diff --git a/.github/workflows/cli-release-tag.yml b/.github/workflows/cli-release-tag.yml
index 744a99c4..570d83a3 100644
--- a/.github/workflows/cli-release-tag.yml
+++ b/.github/workflows/cli-release-tag.yml
@@ -70,4 +70,67 @@ jobs:
           UV_PUBLISH_TOKEN: ${{ secrets.PYPI_API_TOKEN }}
         run: uv publish
 
+  update-brew-formula:
+    needs: [bump, release]
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+    env:
+      RELEASE_TAG: ${{ needs.bump.outputs.tag }}
+    steps:
+      - name: Checkout main
+        uses: actions/checkout@v4
+        with:
+          ref: main
+
+      - name: Download release binaries
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          mkdir -p bin
+          gh release download "$RELEASE_TAG" --repo "${{ github.repository }}" --dir bin --pattern 'canyonos-*'
+
+      - name: Update version and sha256 in Formula/canyonos.rb
+        run: |
+          python3 - <<'PY'
+          import hashlib
+          import os
+          import pathlib
+          import re
+
+          version = os.environ["RELEASE_TAG"].removeprefix("cli-v")
+          assets = [
+              "canyonos-macos-arm64",
+              "canyonos-macos-x86_64",
+              "canyonos-linux-arm64",
+              "canyonos-linux-x86_64",
+          ]
+
+          formula = pathlib.Path("Formula/canyonos.rb")
+          text = formula.read_text()
+          text = re.sub(r'version ".*"', f'version "{version}"', text, count=1)
+
+          for asset in assets:
+              sha256 = hashlib.sha256(pathlib.Path(f"bin/{asset}").read_bytes()).hexdigest()
+              text = re.sub(
+                  rf'(url "[^"]*/{re.escape(asset)}"\n\s*sha256 ")[a-f0-9]+(")',
+                  rf"\g<1>{sha256}\g<2>",
+                  text,
+              )
+
+          formula.write_text(text)
+          PY
+
+      - name: Commit and push formula update
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git add Formula/canyonos.rb
+          if git diff --cached --quiet; then
+            echo "No formula changes to commit"
+          else
+            git commit -m "brew: update canyonos formula to $RELEASE_TAG [skip release]"
+            git push origin HEAD:main
+          fi
+
 # Calls cli-release.yml directly (a GITHUB_TOKEN tag push won't retrigger it); "[skip release]" is a defensive backstop against the same push looping.
diff --git a/README.md b/README.md
index d097c241..9c3f1a75 100644
--- a/README.md
+++ b/README.md
@@ -57,6 +57,7 @@ Use any of the following package managers to install the canyonos CLI (curl, bre
 curl -fsSL https://raw.githubusercontent.com/CanyonCodeCoreAI/canyoncodecore/main/cli/install.sh | sh
 # OR
 brew tap CanyonCodeCoreAI/canyonos https://github.com/CanyonCodeCoreAI/canyoncodecore
+brew trust --formula CanyonCodeCoreAI/canyonos/canyonos
 brew install canyonos
 # OR
 uv tool install canyonos
@@ -157,7 +158,11 @@ Example Success Message:
 
 ### 4. Sending requests to the workflow
 
-Upon running the deploy command, canyonos automatically generates a REST API endpoint for the workflow. Send requests to this endpoint to trigger the workflow:
+Upon running the deploy command, canyonos automatically generates a REST API endpoint for the workflow. 
+
+For verifying the workflow has been deployed fine, run `canyonos test "A test query"` to send a query through the workflow.
+
+For manually sending requests, use the given endpoint to trigger the workflow:
 
 ```bash
 curl -X POST http://localhost:8000/main \
diff --git a/canyonos_core/cli.py b/canyonos_core/cli.py
index 2064c365..96911467 100644
--- a/canyonos_core/cli.py
+++ b/canyonos_core/cli.py
@@ -51,7 +51,70 @@ def _load_config(config_path):
     import yaml
 
     with open(config_path, "r") as f:
-        return yaml.safe_load(f)
+        config = yaml.safe_load(f)
+    # Everything below here till "return config" is basically just checks to make sure the folder is correct
+    if not isinstance(config, dict):
+        raise RuntimeError(f"Config must contain a YAML mapping: {config_path}")
+    agents = config.get("agents", [])
+    if not isinstance(agents, list) or not all(isinstance(agent, dict) for agent in agents):
+        raise RuntimeError(f"Config `agents` must be a list of mappings: {config_path}")
+    names = [agent.get("name") for agent in agents]
+    if any(not isinstance(name, str) or not name.strip() for name in names):
+        raise RuntimeError(f"Every configured agent must have a non-empty name: {config_path}")
+    names_by_key = {}
+    for name in names:
+        names_by_key.setdefault(name.casefold(), []).append(name)
+    duplicates = sorted(
+        "/".join(group) for group in names_by_key.values() if len(group) > 1
+    )
+    if duplicates:
+        raise RuntimeError(f"Duplicate agent names in {config_path}: {', '.join(duplicates)}")
+
+    for agent in agents:
+        name = agent["name"]
+        provider = agent.get("provider", "local")
+        if not isinstance(provider, str) or provider.casefold() not in {"local", "ec2"}:
+            raise RuntimeError(
+                f"Agent {name} has unsupported provider {provider!r}; use `local` or `EC2`."
+            )
+        agent["provider"] = "EC2" if provider.casefold() == "ec2" else "local"
+
+        replicas = agent.get("replicas", 1)
+        if isinstance(replicas, bool) or not isinstance(replicas, int) or replicas < 1:
+            raise RuntimeError(f"Agent {name} must have a positive integer `replicas` value.")
+        if (
+            agent["provider"] == "local"
+            and agent.get("type", "agent") == "workflow"
+            and replicas > 1
+        ):
+            raise RuntimeError(
+                f"Local workflow {name} cannot use replicas > 1 because every replica "
+                "would publish the same `api_port`."
+            )
+
+        for field in ("host_port", "port", "redis_port", "api_port", "dashboard_port"):
+            value = agent.get(field)
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 65535
+            ):
+                raise RuntimeError(
+                    f"Agent {name} must have an integer `{field}` between 1 and 65535."
+                )
+
+        resources = agent.get("resources", {})
+        if not isinstance(resources, dict):
+            raise RuntimeError(f"Agent {name} `resources` must be a mapping.")
+        for field in ("cpu", "memory", "gpu"):
+            value = resources.get(field)
+            if value is not None and (
+                isinstance(value, bool)
+                or not isinstance(value, (int, float))
+                or value <= 0
+            ):
+                raise RuntimeError(
+                    f"Agent {name} resource `{field}` must be a positive number."
+                )
+    return config
 
 
 def _artifact_prefix(root):
@@ -237,6 +300,17 @@ def _run_build(config_path):
     source_root = os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir
     package_dir = _get_package_dir()
 
+    missing_sources = []
+    for agent in agents:
+        source_field = "workflow_file" if agent.get("type", "agent") == "workflow" else "entrypoint"
+        source_path = agent.get(source_field)
+        if not isinstance(source_path, str) or not source_path:
+            missing_sources.append(f"{agent['name']}: missing `{source_field}`")
+        elif not os.path.isfile(os.path.join(source_root, source_path)):
+            missing_sources.append(f"{agent['name']}: {source_path} not found")
+    if missing_sources:
+        raise RuntimeError("Cannot build configured sources: " + "; ".join(missing_sources))
+
     # -------------------------------------------------------------- #
     #  Step 1: Discover agent YAML files and generate Python stubs    #
     # -------------------------------------------------------------- #
diff --git a/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py b/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
index 751daff0..3ddc169f 100644
--- a/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
+++ b/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
@@ -8,6 +8,7 @@
 
 import logging
 import os
+import socket
 
 from canyonos_core.controller.utils.container_names import container_name
 from canyonos_core.controller.utils.env_file import env_file_args
@@ -19,6 +20,7 @@
 PROVIDER = "local"
 MAX_PORT_ATTEMPTS = 50
 NETWORK = "canyonos-local"
+HOST_GATEWAY = "host.docker.internal"
 _controller = None
 
 
@@ -36,6 +38,16 @@ def validate_config():
     return None
 
 
+def _port_check(host, port):
+    """Preflight check to test if the port (8080) is available, fails immediately if not instead of failing later"""
+    probe_host = HOST_GATEWAY if _is_local_host(host) else host
+    try:
+        with socket.create_connection((probe_host, int(port)), timeout=0.25):
+            return True
+    except OSError:
+        return False
+
+
 def provision_instance(spec, replica_index, next_host_port):
     host = spec.get("host", DEFAULT_HOST)
     host_port = int(spec.get("host_port", spec.get("port", next_host_port(host))))
@@ -73,6 +85,12 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         )
         _require_controller()._run_cmd(["docker", "rm", "-f", runtime_id], host, user)
 
+    if ctrl_type == "workflow" and _port_check(host, spec.get("api_port", 8080)):
+        raise RuntimeError(
+            f"Cannot launch {runtime_id}: workflow api_port "
+            f"{spec.get('api_port', 8080)} is already in use on {host}."
+        )
+
     for attempt in range(MAX_PORT_ATTEMPTS):
         cmd = [
             "docker",
@@ -140,7 +158,8 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
 
         if result.returncode == 0:
             break
-        if "port is already allocated" in (result.stderr or ""):
+        error = (result.stderr or "").lower()
+        if "port is already allocated" in error or "address already in use" in error:
             # `docker run` leaves a `Created`-but-never-started container behind
             # under this name when the port bind fails. Remove it before
             # retrying with a new port, or the retry hits a name conflict
@@ -189,7 +208,12 @@ def terminate_instance(instance):
         instance.get("user"),
     )
     if result.returncode != 0:
-        logger.warning("Failed to remove runtime %s", runtime_id)
+        detail = (result.stderr or result.stdout or "").strip()
+        if "no such container" not in detail.casefold():
+            raise RuntimeError(
+                f"Failed to remove runtime {runtime_id}: "
+                f"{detail or f'exit code {result.returncode}'}"
+            )
 
 
 def routing_endpoint_for(instance):
diff --git a/canyonos_core/controller/global_controller.py b/canyonos_core/controller/global_controller.py
index b6193df9..69e93560 100644
--- a/canyonos_core/controller/global_controller.py
+++ b/canyonos_core/controller/global_controller.py
@@ -96,7 +96,7 @@ def __init__(self, config_path):
         self.controllers = self.config.get("agents", [])
         self.running = False
         self.containers = {}  # name -> [container_name, ...]
-        self.redis_containers = {}  # host -> container_name
+        self.redis_containers = {}  # host -> owned container_name
         self.node_redis = {}  # host -> RedisClient
         self._last_status = {}  # (host, port) -> last known status
         self._last_metrics_poll_time = {}  # (host, port) -> time.time() of last metrics read
@@ -113,48 +113,70 @@ def __init__(self, config_path):
         # Clean up any stale containers from previous runs
         self._cleanup_stale_containers()
 
-        # Launch Redis on each unique node, then write routing table and policies
-        self._launch_redis_containers()
-        write_agent_specs(self.config_path, self.redis)
-        self._write_resource_specs()
-        self._load_and_write_policies()
-        self._write_identity()
-        self.instance_manager.publish_routing_snapshot(self.controllers)
-        logger.info(
-            "Global controller initialized with %d controller(s).",
-            len(self.controllers),
-        )
-
-        # Start background cleanup thread
-        self._cleanup_ready = threading.Event()
-        self._cleanup_thread = threading.Thread(target=self._cleanup_loop, daemon=True)
-        self._cleanup_thread.start()
-
-        # Spawn the OTLP exporter as a separate process (see canyonos/OTLP_Exporter/DESIGN.md),
-        # supervised so it gets restarted if it ever exits unexpectedly.
-        otel_exporter_dir = os.path.join(
-            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
-            "OTLP_Exporter",
-        )
-        otel_exporter_script = os.path.join(otel_exporter_dir, "otel_exporter.py")
         self.process_supervisor = ProcessSupervisor()
+        try:
+            # Launch Redis on each unique node, then write routing table and policies
+            self._launch_redis_containers()
+            write_agent_specs(self.config_path, self.redis)
+            self._write_resource_specs()
+            self._load_and_write_policies()
+            self._write_identity()
+            self.instance_manager.publish_routing_snapshot(self.controllers)
+            logger.info(
+                "Global controller initialized with %d controller(s).",
+                len(self.controllers),
+            )
 
-        # Exporter polls self.OTEL_DESTINATIONS_KEY in Redis each cycle instead of
-        # reading env once, so reload_config() can update it without a restart.
-        destinations = self._otel_destinations(self.config.get("otel", {}))
-        if destinations is not None:
-            self._write_otel_destinations(destinations)
-            self.process_supervisor.register(
-                "otel_exporter", [sys.executable, otel_exporter_script]
+            # Spawn the OTLP exporter as a separately supervised process.
+            otel_exporter_dir = os.path.join(
+                os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                "OTLP_Exporter",
             )
-        else:
-            logger.info("otel.destinations not configured -- no OTel metrics collection will happen.")
+            otel_exporter_script = os.path.join(otel_exporter_dir, "otel_exporter.py")
+
+            # Exporter reads destinations from Redis so reload_config() can update them.
+            destinations = self._otel_destinations(self.config.get("otel", {}))
+            if destinations is not None:
+                self._write_otel_destinations(destinations)
+                self.process_supervisor.register(
+                    "otel_exporter", [sys.executable, otel_exporter_script]
+                )
+            else:
+                logger.info(
+                    "otel.destinations not configured -- no OTel metrics collection will happen."
+                )
+
+            # Initialize the waiting table before the controller or exporter can use it.
+            self._otel_db = otel_db
+            self._otel_db.init_db()
+            self.process_supervisor.start_all()
 
-        # Initialize/migrate the waiting table synchronously before either the GC or
-        # exporter process can access it.
-        self._otel_db = otel_db
-        self._otel_db.init_db()
-        self.process_supervisor.start_all()
+            self._cleanup_ready = threading.Event()
+            self._cleanup_thread = threading.Thread(
+                target=self._cleanup_loop, daemon=True
+            )
+            self._cleanup_thread.start()
+        except Exception:
+            try:
+                self.process_supervisor.terminate_all()
+            except Exception as cleanup_error:
+                logger.warning(
+                    "Failed to stop supervised processes after initialization failure: %s",
+                    cleanup_error,
+                )
+            try:
+                redis_failures = self._stop_redis_containers()
+                if redis_failures:
+                    logger.warning(
+                        "Redis cleanup after initialization failure was incomplete: %s",
+                        "; ".join(redis_failures),
+                    )
+            except Exception as cleanup_error:
+                logger.warning(
+                    "Failed to clean up Redis after initialization failure: %s",
+                    cleanup_error,
+                )
+            raise
 
     # ------------------------------------------------------------------ #
     #  Stale container cleanup                                             #
@@ -213,6 +235,8 @@ def _load_config(config_path):
         GlobalController._load_dotenv(os.path.join(project_root, ".env"))
         with open(config_path, "r") as f:
             config = yaml.safe_load(f)
+        if not isinstance(config, dict):
+            raise RuntimeError(f"Config must contain a YAML mapping: {config_path}")
         if not config.get("project_id"):
             config["project_id"] = GlobalController._assign_new_project_id(config_path)
         config = GlobalController._expand_env_value(config)
@@ -420,7 +444,6 @@ def _launch_redis_containers(self):
 
             if self._redis_container_healthy(container_name, host, user, connect_host, redis_port):
                 logger.info("Reusing existing Redis container %s on %s", container_name, host)
-                self.redis_containers[host] = container_name
             else:
                 if _is_local_host(host):
                     self._run_cmd(["docker", "network", "create", LOCAL_NETWORK], host, user)
@@ -437,31 +460,54 @@ def _launch_redis_containers(self):
                     "redis:alpine",
                 ]
 
-                try:
-                    result = self._run_cmd(cmd, host, user)
-                    if result.returncode == 0:
-                        self.redis_containers[host] = container_name
-                        logger.info(
-                            "Launched Redis container %s on %s:%d",
-                            container_name,
-                            host,
-                            redis_port,
+                launch_attempts = 3
+                launch_error = None
+                for attempt in range(1, launch_attempts + 1):
+                    try:
+                        result = self._run_cmd(cmd, host, user)
+                        if result.returncode == 0:
+                            self.redis_containers[host] = container_name
+                            logger.info(
+                                "Launched Redis container %s on %s:%d",
+                                container_name,
+                                host,
+                                redis_port,
+                            )
+                            break
+                        launch_error = result.stderr.strip() or (
+                            f"docker run exited with code {result.returncode}"
                         )
-                    else:
-                        logger.critical(
-                            "Failed to launch Redis on %s: %s",
+                    except Exception as e:
+                        launch_error = str(e)
+                    logger.warning(
+                        "Redis launch attempt %d/%d failed on %s: %s",
+                        attempt,
+                        launch_attempts,
+                        host,
+                        launch_error,
+                    )
+                    # A `docker run` that fails to bind still leaves a Created
+                    # container holding this name, so every retry would fail as
+                    # "name is already in use" and bury the real error.
+                    try:
+                        self._run_cmd(["docker", "rm", "-f", container_name], host, user)
+                    except Exception as e:
+                        logger.warning(
+                            "Could not clear Redis container name %s on %s before retrying: %s",
+                            container_name,
                             host,
-                            result.stderr.strip(),
+                            e,
                         )
-                        sys.exit(1)
-                except FileNotFoundError:
-                    logger.critical(
-                        "Docker is not installed or not in PATH. Cannot launch Redis."
+                else:
+                    rollback_failures = self._stop_redis_containers()
+                    message = (
+                        f"Failed to launch Redis on {host} after {launch_attempts} "
+                        f"attempts: {launch_error}"
                     )
-                    sys.exit(1)
-                except Exception as e:
-                    logger.critical("Failed to launch Redis on %s: %s", host, e)
-                    sys.exit(1)
+                    if rollback_failures:
+                        message += "; rollback incomplete: " + "; ".join(rollback_failures)
+                    logger.critical(message)
+                    raise RuntimeError(message)
 
             # Create a RedisClient for this node
             redis_client = RedisClient(host=connect_host, port=redis_port)
@@ -472,10 +518,15 @@ def _launch_redis_containers(self):
         if "localhost" in self.node_redis:
             self.redis = self.node_redis["localhost"]
 
-        logger.info("Redis launched on %d node(s).", len(self.redis_containers))
+        logger.info(
+            "Redis ready on %d node(s); %d owned by this controller.",
+            len(self.node_redis),
+            len(self.redis_containers),
+        )
 
     def _stop_redis_containers(self):
-        """Stop and remove all launched Redis containers."""
+        """Stop and remove all Redis containers owned by this controller."""
+        failures = []
         nodes = {}
         for ctrl in self.controllers:
             if ctrl.get("provider", "local").upper() == "EC2":
@@ -484,17 +535,25 @@ def _stop_redis_containers(self):
             redis_port = ctrl.get("redis_port", 6379)
             for host, _port in self._get_replica_placements(ctrl):
                 nodes.setdefault(host, {"user": user, "redis_port": redis_port})
-        for host, container_name in self.redis_containers.items():
+        for host, container_name in list(self.redis_containers.items()):
             user = nodes.get(host, {}).get("user")
-            try:
-                self._run_cmd(["docker", "stop", container_name], host, user)
-                self._run_cmd(["docker", "rm", container_name], host, user)
+            errors = []
+            for action in ("stop", "rm"):
+                try:
+                    result = self._run_cmd(["docker", action, container_name], host, user)
+                    if result.returncode != 0 and "no such container" not in (result.stderr or "").lower():
+                        errors.append(f"docker {action} exited with code {result.returncode}: {(result.stderr or '').strip()}")
+                except Exception as e:
+                    errors.append(f"docker {action}: {e}")
+            if errors:
+                detail = f"Redis {container_name} on {host}: " + "; ".join(errors)
+                failures.append(detail)
+                logger.warning("Failed to fully stop %s", detail)
+            else:
+                self.redis_containers.pop(host, None)
+                self.node_redis.pop(host, None)
                 logger.info("Stopped Redis %s on %s", container_name, host)
-            except Exception as e:
-                logger.warning("Failed to stop Redis %s: %s", container_name, e)
-
-        self.redis_containers.clear()
-        self.node_redis.clear()
+        return failures
 
     # ------------------------------------------------------------------ #
     #  Startup health check                                               #
@@ -548,6 +607,14 @@ def _wait_for_healthy(self, timeout=30, interval=2):
                     instance["host_port"],
                     timeout,
                 )
+            names = ", ".join(
+                f"{instance['agent_name']} ({instance['host']}:{instance['host_port']})"
+                for instance in pending
+            )
+            raise RuntimeError(
+                f"{len(pending)} controller replica(s) failed to become healthy "
+                f"within {timeout}s: {names}"
+            )
 
     # ------------------------------------------------------------------ #
     #  Polling loop                                                       #
@@ -845,18 +912,23 @@ def _run_cmd(self, cmd, host, user=None):
             subprocess.CompletedProcess
         """
         is_local = _is_local_host(host)
-        if is_local:
-            return subprocess.run(cmd, capture_output=True, text=True, timeout=180)
-
-        remote_cmd = " ".join(cmd)
-        if cmd and cmd[0] == "docker":
-            remote_cmd = f"sudo {remote_cmd}"
-        return subprocess.run(
-            self._ssh_args(host, user) + [remote_cmd],
-            capture_output=True,
-            text=True,
-            timeout=180,
-        )
+        try:
+            if is_local:
+                return subprocess.run(cmd, capture_output=True, text=True, timeout=180)
+
+            remote_cmd = " ".join(cmd)
+            if cmd and cmd[0] == "docker":
+                remote_cmd = f"sudo {remote_cmd}"
+            return subprocess.run(
+                self._ssh_args(host, user) + [remote_cmd],
+                capture_output=True,
+                text=True,
+                timeout=180,
+            )
+        except subprocess.TimeoutExpired:
+            raise RuntimeError(f"Command timed out after 180s on {host}: {' '.join(cmd)}") from None
+        except OSError as e:
+            raise RuntimeError(f"Could not run command on {host}: {e}") from None
 
     def _push_file(self, local_path, remote_path, host, user=None):
         """
@@ -877,15 +949,20 @@ def _push_file(self, local_path, remote_path, host, user=None):
         """
         quoted = shlex.quote(remote_path)
         remote_cmd = f"umask 077; rm -f {quoted}; cat > {quoted}"
-        with open(local_path, "rb") as f:
-            result = subprocess.run(
-                self._ssh_args(host, user) + [remote_cmd],
-                stdin=f,
-                capture_output=True,
-                text=True,
-                timeout=180,
-                check=False,
-            )
+        try:
+            with open(local_path, "rb") as f:
+                result = subprocess.run(
+                    self._ssh_args(host, user) + [remote_cmd],
+                    stdin=f,
+                    capture_output=True,
+                    text=True,
+                    timeout=180,
+                    check=False,
+                )
+        except subprocess.TimeoutExpired:
+            raise RuntimeError(f"Copying {local_path} to {host} timed out after 180s") from None
+        except OSError as e:
+            raise RuntimeError(f"Could not copy {local_path} to {host}: {e}") from None
         if result.returncode != 0:
             raise RuntimeError(
                 f"Failed to copy {local_path} to {host}:{remote_path}: "
@@ -917,13 +994,19 @@ def launch_docker_agents(self):
 
     def _stop_docker_agents(self):
         """Stop and remove all launched runtimes."""
+        failures = []
         for instance in self.instance_manager.list_instances():
-            self.instance_manager.remove_instance(
-                self.instance_manager._instance_id_from_record(instance)
-            )
+            instance_id = self.instance_manager._instance_id_from_record(instance)
+            try:
+                self.instance_manager.remove_instance(instance_id)
+            except Exception as e:
+                failures.append(f"agent {instance_id}: {e}")
+                logger.warning("Failed to stop agent %s: %s", instance_id, e)
 
-        self.containers.clear()
-        logger.info("All Docker containers stopped.")
+        if not failures:
+            self.containers.clear()
+            logger.info("All Docker containers stopped.")
+        return failures
 
     # ------------------------------------------------------------------ #
     #  Shutdown                                                           #
@@ -932,17 +1015,28 @@ def _stop_docker_agents(self):
     def cleanup(self):
         """Full cleanup — stop all containers and Redis, called on exit."""
         if not self.running and not self.containers and not self.redis_containers:
-            return  # Already cleaned up
+            return []  # Already cleaned up
         logger.info("Cleaning up all resources...")
-        self.stop()
+        return self.stop()
 
     def stop(self):
-        """Gracefully shut down the daemon and all agent processes."""
+        """Gracefully shut down the daemon and all agent processes.
+
+        Returns what it could not remove rather than raising: this runs from a
+        signal handler and again from atexit, where an exception would skip the
+        handler's own exit and then repeat on the way out.
+        """
         self.running = False
-        self._stop_docker_agents()
-        self._stop_redis_containers()
-        self.process_supervisor.terminate_all()
-        logger.info("Global controller shut down.")
+        failures = (self._stop_docker_agents() or []) + (self._stop_redis_containers() or [])
+        try:
+            self.process_supervisor.terminate_all()
+        except Exception as e:
+            failures.append(f"OTel exporter: {e}")
+        if failures:
+            logger.error("Cleanup incomplete:\n- %s", "\n- ".join(failures))
+        else:
+            logger.info("Global controller shut down.")
+        return failures
 
 
 if __name__ == "__main__":
diff --git a/canyonos_core/controller/instance_manager.py b/canyonos_core/controller/instance_manager.py
index a389d143..ed1179f4 100644
--- a/canyonos_core/controller/instance_manager.py
+++ b/canyonos_core/controller/instance_manager.py
@@ -7,6 +7,7 @@
 """
 
 import json
+import logging
 import os
 import uuid
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -15,6 +16,7 @@
 from canyonos_core.controller.utils import container_names
 
 DEFAULT_HOST_PORT_START = 8000
+logger = logging.getLogger(__name__)
 
 
 class InstanceManager:
@@ -31,7 +33,22 @@ def redis(self):
         return self._redis or self.controller.redis
 
     def ensure_instances(self, agent_specs):
-        self._agent_specs = list(agent_specs)
+        self._agent_specs = []
+        for original in agent_specs:
+            agent_spec = dict(original)
+            provider = agent_spec.get("provider", "local")
+            if not isinstance(provider, str) or provider.casefold() not in {
+                "local",
+                "ec2",
+            }:
+                raise RuntimeError(
+                    f"Agent {agent_spec.get('name', '<unnamed>')} has unsupported "
+                    f"provider {provider!r}; use `local` or `EC2`."
+                )
+            agent_spec["provider"] = (
+                "EC2" if provider.casefold() == "ec2" else "local"
+            )
+            self._agent_specs.append(agent_spec)
         instances = []
         existing = []
         jobs = []
@@ -52,8 +69,11 @@ def ensure_instances(self, agent_specs):
                 instance = self.redis.hgetall(key)
 
                 if instance and instance.get("runtime_id"):
-                    existing.append((agent_name, instance_id, instance))
-                    continue
+                    if self._runtime_is_running(instance):
+                        existing.append((agent_name, instance_id, instance))
+                        continue
+                    self._destroy_runtime(instance)
+                    self._discard_instance_record(instance_id, instance)
 
                 reserved_port = None
                 if provider == "local":
@@ -106,11 +126,42 @@ def _provision_one(self, job):
         provisioned = runtime.provision_instance(
             agent_spec, replica_index, next_host_port
         )
-        agent_id = uuid.uuid4().hex
-        instance = runtime.bootstrap_instance(provisioned, agent_spec, replica_index, agent_id)
-        instance["agent_id"] = agent_id
-        self._write_instance(instance)
-        return instance
+        runtime_id = provisioned.get("runtime_id")
+        if runtime_id:
+            self._track_runtime(job["agent_name"], runtime_id)
+
+        instance = provisioned
+        try:
+            agent_id = uuid.uuid4().hex
+            instance = runtime.bootstrap_instance(
+                provisioned, agent_spec, replica_index, agent_id
+            )
+            instance["agent_id"] = agent_id
+            self._write_instance(instance)
+            return instance
+        except Exception:
+            try:
+                runtime.terminate_instance(instance)
+            except Exception as cleanup_error:
+                logger.warning(
+                    "Failed to clean up runtime %s after provisioning failed: %s",
+                    runtime_id,
+                    cleanup_error,
+                )
+            else:
+                self._untrack_runtime(job["agent_name"], runtime_id)
+            try:
+                self.redis.delete(f"agent_instance:{job['instance_id']}")
+                self.redis.srem(
+                    f"agent:{job['agent_name']}:instances", job["instance_id"]
+                )
+            except Exception as cleanup_error:
+                logger.warning(
+                    "Failed to remove the partial instance record for %s: %s",
+                    job["instance_id"],
+                    cleanup_error,
+                )
+            raise
 
     def _write_instance(self, instance):
         key = self._instance_key(
@@ -193,6 +244,44 @@ def _track_runtime(self, agent_name, runtime_id):
         if runtime_id not in containers:
             containers.append(runtime_id)
 
+    def _untrack_runtime(self, agent_name, runtime_id):
+        self.controller.containers[agent_name] = [
+            tracked
+            for tracked in self.controller.containers.get(agent_name, [])
+            if tracked != runtime_id
+        ]
+
+    def _runtime_is_running(self, instance):
+        runtime_id = instance.get("runtime_id")
+        host = instance.get("host")
+        if not runtime_id or not host:
+            return False
+
+        provider = instance.get("provider", "local")
+        user = instance.get("user")
+        if provider.casefold() == "ec2":
+            runtime_id = runtime_id.rsplit("--", 1)[0]
+            user = user or self.controller.config.get("ec2", {}).get("ssh_user")
+
+        try:
+            result = self.controller._run_cmd(
+                ["docker", "inspect", "-f", "{{.State.Running}}", runtime_id],
+                host,
+                user,
+            )
+        except RuntimeError as e:
+            # _run_cmd raises when the host can't be reached at all. This
+            # answers whether a record is reusable, and an unreachable host
+            # isn't; let the re-provision that follows report the real failure.
+            logger.warning("Could not inspect %s on %s: %s", runtime_id, host, e)
+            return False
+        return result.returncode == 0 and result.stdout.strip() == "true"
+
+    def _discard_instance_record(self, instance_id, instance):
+        self.redis.delete(f"agent_instance:{instance_id}")
+        self.redis.srem(f"agent:{instance['agent_name']}:instances", instance_id)
+        self._untrack_runtime(instance["agent_name"], instance["runtime_id"])
+
     def _next_host_port(self, host, key, agent_name, provider, replica_index):
         used = {
             int(instance["host_port"])
@@ -232,10 +321,13 @@ def container_name(self, agent_spec, replica_index):
         return container_names.container_name(agent_spec["name"], replica_index)
 
     def _provider_runtime(self, provider):
-        if provider.upper() == "EC2":
+        normalized = provider.casefold()
+        if normalized == "ec2":
             from canyonos_core.controller.cloud_provider_logic.EC2 import _runtime as runtime
-        else:
+        elif normalized == "local":
             runtime = local_runtime
+        else:
+            raise RuntimeError(f"Unsupported provider {provider!r}; use `local` or `EC2`.")
         runtime._controller = self.controller
         return runtime
 
diff --git a/canyonos_core/controller/local_controller.py b/canyonos_core/controller/local_controller.py
index 3ee93292..cbb6191a 100644
--- a/canyonos_core/controller/local_controller.py
+++ b/canyonos_core/controller/local_controller.py
@@ -10,7 +10,7 @@
 import threading
 import time
 import importlib.util
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, wait
 
 import grpc
 import psutil
@@ -56,6 +56,7 @@
 ROUTING_ENDPOINTS_KEY = "routing_table:endpoints"
 ROUTING_STATEFUL_KEY = "routing_table:stateful"
 POLICY_RULES_KEY = "policy:rules"
+EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 5
 
 
 class LocalController(object):
@@ -79,12 +80,12 @@ def __init__(self, port=50051):
         # soon as it arrives via WriteResult (see _fan_out_to_consumers).
         self.servicer.on_result = self._fan_out_to_consumers
 
-        # Connect to Redis and report healthy status
+        # Connect to Redis. Readiness is published only after a configured
+        # agent has loaded successfully.
         redis_host = os.environ.get("CANYONOS_REDIS_HOST", "localhost")
         redis_port = int(os.environ.get("CANYONOS_REDIS_PORT", 6379))
         self.redis = RedisClient(host=redis_host, port=redis_port)
         self._status_key = f"controller:{self.agent_host}:{self.public_port}:status"
-        self.redis.set(self._status_key, "healthy")
 
         # Set once by InstanceManager when this replica was provisioned; read back
         # here so completed requests can be stamped with which replica ran them.
@@ -99,7 +100,6 @@ def __init__(self, port=50051):
         psutil.cpu_percent(interval=None)  # prime so the first real reading isn't 0.0
         self._metrics_stop_event = threading.Event()
         self._metrics_thread = threading.Thread(target=self._metrics_loop, daemon=True)
-        self._metrics_thread.start()
 
         # Cache for gRPC stubs to remote controllers
         self._remote_channels = {}  # endpoint -> grpc.Channel
@@ -113,21 +113,35 @@ def __init__(self, port=50051):
         # that need to be routed through the same controller's request queue.
         max_instances = int(os.environ.get("CANYONOS_MAX_AGENT_INSTANCES", 8))
         self._executor = ThreadPoolExecutor(max_workers=max_instances)
+        self._executor_futures = {}
+        self._executor_futures_lock = threading.Lock()
 
         # Start the LLM proxy alongside the agent in this container. Bedrock
         # calls are routed to it via AWS_ENDPOINT_URL_BEDROCK_RUNTIME (injected
         # by the runtime), and it writes token/cost telemetry to Redis.
         self._proxy_process = self._start_llm_proxy(redis_host, redis_port)
 
+        # After the proxy, because an agent constructor may build an LLM client
+        # against it. Workflow containers intentionally run a routing-only local
+        # controller without these variables; agent containers set both, and
+        # must not advertise readiness if constructing the agent failed.
+        self.agent = self._load_agent()
+        if (self.agent_name or self.agent_file) and self.agent is None:
+            self.redis.set(self._status_key, "failed")
+            self.server.stop(0)
+            raise RuntimeError(
+                f"Failed to load configured agent {self.agent_name or self.agent_file}."
+            )
+
+        self.redis.set(self._status_key, "healthy")
+        self._metrics_thread.start()
+
         logger.info(
             "Local controller initialized at %s (max_agent_instances=%d), reported healthy to Redis.",
             self._my_endpoint,
             max_instances,
         )
 
-        # Load the agent class dynamically
-        self.agent = self._load_agent()
-
     def _start_llm_proxy(self, redis_host, redis_port):
         """Start the LLM proxy as a subprocess in this container (127.0.0.1:8081).
 
@@ -454,7 +468,7 @@ def _process_request(self, data):
 
         if endpoint == self._my_endpoint:
             submitted_at = time.time()
-            self._executor.submit(
+            executor_future = self._executor.submit(
                 self._execute_locally,
                 service,
                 function,
@@ -466,6 +480,9 @@ def _process_request(self, data):
                 parent,
                 created_at,
             )
+            with self._executor_futures_lock:
+                self._executor_futures[executor_future] = future_id
+            executor_future.add_done_callback(self._forget_executor_future)
         else:
             # Register the target as a consumer for any Future args
             # so results get pushed to its Redis via WriteResult.
@@ -805,12 +822,28 @@ def _fan_out_to_consumers(self, future_id, result=None, failed=0, error_message=
     #  Shutdown                                                            #
     # ------------------------------------------------------------------ #
 
+    def _forget_executor_future(self, executor_future):
+        with self._executor_futures_lock:
+            self._executor_futures.pop(executor_future, None)
+
     def stop(self):
         """Gracefully shut down the server."""
         logger.info("Shutting down local controller...")
         self._metrics_stop_event.set()
         self._metrics_thread.join(timeout=2)
-        self._executor.shutdown(wait=True)
+        self._executor.shutdown(wait=False, cancel_futures=True)
+        with self._executor_futures_lock:
+            outstanding = dict(self._executor_futures)
+        if outstanding:
+            _, unfinished = wait(
+                outstanding, timeout=EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS
+            )
+            if unfinished:
+                future_ids = sorted(str(outstanding[future]) for future in unfinished)
+                logger.warning(
+                    "Executor shutdown timed out with requests still running: %s",
+                    ", ".join(future_ids),
+                )
         self.redis.set(self._status_key, "stopped")
         self.server.stop(0)
 
diff --git a/canyonos_core/server.py b/canyonos_core/server.py
index c1e4b48e..7b86ac70 100644
--- a/canyonos_core/server.py
+++ b/canyonos_core/server.py
@@ -2,6 +2,7 @@
 import signal
 import subprocess
 import sys
+import threading
 
 import yaml
 from flask import Flask, jsonify, request
@@ -16,9 +17,11 @@
 WORKSPACE_DIR = "/workspace"
 
 DEFAULT_API_PORT = 8080
+CLEAN_TIMEOUT_SECONDS = 45
 
 _gc_process = None
 _config_path = None
+_gc_lock = threading.Lock()
 
 
 def _gc_running():
@@ -34,9 +37,6 @@ def new_project():
 def deploy():
     global _gc_process, _config_path
 
-    if _gc_running():
-        return jsonify({"error": "already running"}), 409
-
     data = request.get_json(force=True, silent=True) or {}
     # Resolved with canyonos' own artifact-layout rule rather than a second copy
     # of it, so a `.car` project works when the client sends no config_path.
@@ -56,25 +56,43 @@ def deploy():
     # Controller. cwd is the workspace so build outputs land alongside the
     # project files and the controller finds them. Build+deploy output streams
     # to the container logs, which `canyonos deploy` tails.
-    _gc_process = subprocess.Popen(
-        [sys.executable, "-m", "canyonos_core.cli", "deploy", "-c", config_path],
-        cwd=WORKSPACE_DIR,
-    )
-    _config_path = full_path
-    return jsonify({"status": "started", "pid": _gc_process.pid}), 200
+    with _gc_lock:
+        if _gc_running():
+            return jsonify({"error": "already running"}), 409
+        _gc_process = subprocess.Popen(
+            [sys.executable, "-m", "canyonos_core.cli", "deploy", "-c", config_path],
+            cwd=WORKSPACE_DIR,
+        )
+        _config_path = full_path
+        return jsonify({"status": "started", "pid": _gc_process.pid}), 200
 
 
 @app.route("/clean", methods=["POST"])
 def clean():
     global _gc_process
 
-    if not _gc_running():
-        return jsonify({"error": "not running"}), 409
-
-    _gc_process.send_signal(signal.SIGTERM)
-    _gc_process.wait()
-    _gc_process = None
-    return jsonify({"status": "stopped"}), 200
+    with _gc_lock:
+        if not _gc_running():
+            return jsonify({"error": "not running"}), 409
+
+        process = _gc_process
+        process.send_signal(signal.SIGTERM)
+        try:
+            process.wait(timeout=CLEAN_TIMEOUT_SECONDS)
+        except subprocess.TimeoutExpired:
+            return jsonify(
+                {
+                    "error": (
+                        "Global Controller did not stop within "
+                        f"{CLEAN_TIMEOUT_SECONDS} seconds"
+                    )
+                }
+            ), 504
+        return_code = process.returncode
+        _gc_process = None
+        if return_code:
+            return jsonify({"error": f"Global Controller stopped with exit code {return_code}"}), 500
+        return jsonify({"status": "stopped"}), 200
 
 
 @app.route("/status", methods=["GET"])
diff --git a/cli/canyonos/dashboard_stack.py b/cli/canyonos/dashboard_stack.py
index d3fa6023..19d1f667 100644
--- a/cli/canyonos/dashboard_stack.py
+++ b/cli/canyonos/dashboard_stack.py
@@ -372,15 +372,18 @@ def _cleanup(stack: DashboardStack, manifest: Path) -> None:
 def _dashboard_compose_command(*args: str) -> bool:
     """Run a `docker compose` subcommand against the dashboard stack from the current project.
 
-    False (no-op) if the dashboard was never started from here -- there's no
-    `.env` for `--env-file` to point at, so there's nothing to stop/tear down.
+    A missing dashboard is already the desired end state and returns True.
+    False is reserved for a Compose command that actually failed.
     """
     stack = DashboardStack(state_dir=_state_dir(), project_dir=Path.cwd())
     if not stack.env_path.is_file():
-        return False
+        return True
     manifest_resource = importlib.resources.files("canyonos").joinpath("dashboard.compose.yml")
-    with importlib.resources.as_file(manifest_resource) as manifest:
-        result = _run([*_compose_argv(stack, manifest), *args])
+    try:
+        with importlib.resources.as_file(manifest_resource) as manifest:
+            result = _run([*_compose_argv(stack, manifest), *args])
+    except OSError:
+        return False
     return result.returncode == 0
 
 
diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py
index 5cab56cb..ca522392 100644
--- a/cli/canyonos/deploy.py
+++ b/cli/canyonos/deploy.py
@@ -12,7 +12,8 @@
 
 Once the deploy's logs report the workflow is actually up, `canyonos serve`
 is kicked off automatically so the local dashboard is ready without an extra
-manual step.
+manual step, and the log tail itself stops -- `canyonos logs` re-attaches
+to it on demand.
 """
 
 import json
@@ -40,6 +41,7 @@
 from canyonos.gc import GCError, deploy_status, post_deploy, workflow_endpoints
 from canyonos.theme import GREEN, WHITE
 from canyonos.init import load_state, run_init
+from canyonos.quit import run_quit
 from canyonos.serve import serve_dashboard
 from canyonos.sync import run_sync
 
@@ -47,6 +49,7 @@
 # wait out the real poll/grace windows.
 _STATUS_POLL_SECONDS = 2.0
 _REVEAL_GRACE_SECONDS = 30.0
+_STARTUP_TIMEOUT_SECONDS = 30 * 60
 
 # Substrings that mean the in-container deploy hit something fatal. `WARNING:` is
 # deliberately absent: the OTel-not-configured notice and stub_generator's
@@ -164,40 +167,60 @@ def run_deploy(config_path=None, serve=True, verbose=False, quiet=False, extra_e
 
     run_init(banner=banner, extra_env=extra_env)
 
-    # Copy the current project into the container before building/deploying.
-    if not run_sync():
-        raise RuntimeError("Could not sync the project into the container.")
+    # Non-empty once the workflow has reported ready; see the handler below.
+    ready = []
+    try:
+        # Copy the current project into the container before building/deploying.
+        if not run_sync():
+            raise RuntimeError("Could not sync the project into the container.")
 
-    state = load_state()
+        state = load_state()
 
-    # Read for display only -- canyonos resolves the path it actually deploys.
-    api_port = workflow_api_port(config_path or default_config_path())
+        # Read for display only -- canyonos resolves the path it actually deploys.
+        api_port = workflow_api_port(config_path or default_config_path())
 
-    # Checked here, after run_init() has already torn down any previous deploy,
-    # so a still-live prior run doesn't read as an unrelated conflict.
-    if api_port is not None and port_in_use(api_port):
-        raise RuntimeError(
-            f"Port {api_port} is already in use, and the workflow needs it. Free it "
-            f"or change `api_port` in {config_path or default_config_path()}."
-        )
+        # Checked here, after run_init() has already torn down any previous deploy,
+        # so a still-live prior run doesn't read as an unrelated conflict.
+        if api_port is not None and port_in_use(api_port):
+            raise RuntimeError(
+                f"Port {api_port} is already in use, and the workflow needs it. Free it "
+                f"or change `api_port` in {config_path or default_config_path()}."
+            )
 
-    try:
-        post_deploy(state["port"], config_path)
-    except GCError as e:
-        raise RuntimeError(str(e)) from None
-
-    if quiet:
-        # Still bring the dashboard up so anything reachable only through its
-        # LLM proxy (e.g. a guardrail calling the OpenAI SDK directly) works
-        # under `canyonos test` too -- just skip the log-tail/summary UI.
-        if serve:
-            _start_dashboard()
+        try:
+            post_deploy(state["port"], config_path)
+        except GCError as e:
+            raise RuntimeError(str(e)) from None
+
+        if quiet:
+            # Still bring the dashboard up so anything reachable only through its
+            # LLM proxy (e.g. a guardrail calling the OpenAI SDK directly) works
+            # under `canyonos test` too -- just skip the log-tail/summary UI.
+            if serve:
+                _start_dashboard()
+            return state
+
+        _stream_logs_and_autoserve(
+            state,
+            api_port,
+            config_path or default_config_path(),
+            serve=serve,
+            verbose=verbose,
+            on_ready=ready.append,
+        )
         return state
-
-    _stream_logs_and_autoserve(
-        state, api_port, config_path or default_config_path(), serve=serve, verbose=verbose
-    )
-    return state
+    except (Exception, KeyboardInterrupt):
+        # Only a deploy that never came up is torn down. Past that point the
+        # workflow is live and serving, and anything that fails afterwards is
+        # a reporting problem, not a reason to take the stack down.
+        if not ready:
+            try:
+                run_quit()
+            except Exception as teardown_error:
+                # Otherwise an unexpected teardown failure would replace the
+                # real deploy error below instead of just supplementing it.
+                ui.fail(f"Teardown after failed deploy also failed: {teardown_error}")
+        raise
 
 
 def _display_host(host):
@@ -286,12 +309,7 @@ def _summary_body(dashboard_url, targets, config_path):
 
 
 def print_deploy_summary(dashboard_url, targets, config_path):
-    """The one screen printed once everything is up: dashboard and workflow endpoints.
-
-    Under `-v` it is printed again on exit, because the log tail continues
-    afterwards and would otherwise scroll it out of sight. Quiet mode prints
-    nothing after it, so once is enough.
-    """
+    """The one screen printed once everything is up: dashboard and workflow endpoints."""
     ui.blank()
     ui.panel(
         Panel(
@@ -315,43 +333,43 @@ def _start_dashboard():
         return None
 
 
-def _deploy_summary(state, api_port, config_path, serve):
+def _deploy_summary(state, api_port, config_path, serve, on_ready):
+    on_ready(True)
     summary = (
         _start_dashboard() if serve else None,
         workflow_targets(state["port"], api_port),
         config_path,
     )
     print_deploy_summary(*summary)
-    ui.hint("Tailing logs now, press Ctrl+C to stop. Run `canyonos stop` to stop the workflow.")
+    ui.hint(
+        "Run `canyonos logs` to view logs, or `canyonos stop` to stop the workflow."
+    )
     return summary
 
 
-def _interrupted(summary=None):
+def _interrupted():
     ui.blank()
     ui.say("Stopped monitoring log stream. Run `canyonos stop` to stop the deploy.")
     ui.hint("To resubscribe to log stream run `canyonos logs`.")
-    if summary is not None:
-        print_deploy_summary(*summary)
-
 
-def _tail_verbose(stream, state, api_port, config_path, serve):
-    """Every log line, verbatim -- what `-v` restores.
 
-    Ctrl+C reprints the summary here but not in quiet mode: only this tail keeps
-    printing past it, so only here has it scrolled out of sight.
-    """
-    summary = None
-    try:
-        for line in stream:
-            print(line, end="")
-            # Logged exactly once, right after the workflow finishes coming up.
-            if summary is None and "Global controller started, polling every" in line:
-                summary = _deploy_summary(state, api_port, config_path, serve)
-    except KeyboardInterrupt:
-        _interrupted(summary)
+def _tail_verbose(lines, state, api_port, config_path, serve, on_ready):
+    """Every log line, verbatim, until the workflow is up -- what `-v` restores."""
+    deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS
+    for line in _drain(lines, state, deadline=deadline, hide_status_requests=False):
+        print(line, end="")
+        # Logged exactly once, right after the workflow finishes coming up.
+        if "Global controller started, polling every" in line:
+            return _deploy_summary(state, api_port, config_path, serve, on_ready)
+
+    raise RuntimeError(
+        "Deploy stopped or timed out before the workflow became ready. "
+        "Automatic cleanup will be attempted; rerun with `canyonos deploy -v` "
+        "for full logs."
+    )
 
 
-def _tail_quiet(lines, state, api_port, config_path, serve):
+def _tail_quiet(lines, state, api_port, config_path, serve, on_ready):
     """Only the phase transitions, until the workflow is up or something fails.
 
     Nothing is echoed raw: the buildx transcript, canyonos' bare prints and grpc's
@@ -368,7 +386,8 @@ def _tail_quiet(lines, state, api_port, config_path, serve):
     # spinner is drawn, and on the way out of a Ctrl+C, so the cursor is restored.
     # A nested spinner wouldn't raise, it would silently render nothing.
     with ui.status("Starting build...") as spinner:
-        for line in _drain(lines, state):
+        deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS
+        for line in _drain(lines, state, deadline=deadline):
             recent.append(line)
             message, done, is_error = tracker.feed(line)
             if is_error:
@@ -381,14 +400,28 @@ def _tail_quiet(lines, state, api_port, config_path, serve):
             if "Global controller started, polling every" in line:
                 summary_line, all_ready = tracker.agents_ready_message()
                 (ui.ok if all_ready else ui.warn)(summary_line)
+                if not all_ready:
+                    ui.fail("Deploy failed.")
+                    ui.blank()
+                    for buffered in recent:
+                        print(buffered, end="")
+                    ui.blank()
+                    raise RuntimeError(
+                        f"Deploy is incomplete: {summary_line.lower()}. "
+                        "The failed deployment will be cleaned up."
+                    )
                 reached_up_marker = True
                 break
 
     if reached_up_marker:
-        return _deploy_summary(state, api_port, config_path, serve)
+        return _deploy_summary(state, api_port, config_path, serve, on_ready)
 
     _reveal_failure(lines, recent, state)
-    return None
+    raise RuntimeError(
+        "Deploy stopped or timed out before the workflow became ready. "
+        "Automatic cleanup will be attempted; rerun with `canyonos deploy -v` "
+        "for full logs."
+    )
 
 
 def _queued_lines(stream):
@@ -401,15 +434,19 @@ def _queued_lines(stream):
     lines = queue.Queue()
 
     def read():
-        for line in stream:
-            lines.put(line)
-        lines.put(None)
+        try:
+            for line in stream:
+                lines.put(line)
+        except OSError as e:
+            lines.put(e)
+        finally:
+            lines.put(None)
 
     threading.Thread(target=read, daemon=True).start()
     return lines
 
 
-def _drain(lines, state, deadline=None):
+def _drain(lines, state, deadline=None, hide_status_requests=True):
     """Yield log lines until the stream ends, the deploy dies, or `deadline` passes.
 
     The container's /status is polled on the read timeout rather than per line,
@@ -429,9 +466,11 @@ def _drain(lines, state, deadline=None):
             continue
         if line is None:
             return
+        if isinstance(line, OSError):
+            raise RuntimeError(f"Could not read Global Controller logs: {line}") from None
         misses = 0
         # Otherwise the container logs its own polling into the stream being read.
-        if "GET /status HTTP/1.1" not in line:
+        if not hide_status_requests or "GET /status HTTP/1.1" not in line:
             yield line
 
 
@@ -466,10 +505,10 @@ def _reveal_failure(lines, recent, state):
     ui.hint("Run `canyonos deploy -v` or `canyonos logs` for the full container log.")
 
 
-def _stream_logs_and_autoserve(state, api_port, config_path, serve=True, verbose=False):
-    """Tail the GC container's logs, and once they show the workflow is up,
-    start the dashboard (unless disabled via `serve=False`) and print where
-    everything lives. Log tailing continues afterwards.
+def _stream_logs_and_autoserve(state, api_port, config_path, serve, verbose, on_ready):
+    """Tail the GC container's logs until the workflow is up, then start the
+    dashboard (unless disabled via `serve=False`), print where everything
+    lives, and stop tailing.
     """
     process = subprocess.Popen(
         ["docker", "logs", "-f", state["container_id"]],
@@ -479,15 +518,11 @@ def _stream_logs_and_autoserve(state, api_port, config_path, serve=True, verbose
         bufsize=1,
     )
     try:
-        if verbose:
-            _tail_verbose(process.stdout, state, api_port, config_path, serve)
-            return
         lines = _queued_lines(process.stdout)
-        if _tail_quiet(lines, state, api_port, config_path, serve) is not None:
-            # Quiet mode stays attached after the summary so Ctrl+C means the
-            # same thing in both modes -- it just swallows what arrives.
-            while lines.get() is not None:
-                pass
+        if verbose:
+            _tail_verbose(lines, state, api_port, config_path, serve, on_ready)
+        else:
+            _tail_quiet(lines, state, api_port, config_path, serve, on_ready)
     except KeyboardInterrupt:
         _interrupted()
     finally:
diff --git a/cli/canyonos/docker_cmd.py b/cli/canyonos/docker_cmd.py
new file mode 100644
index 00000000..9e6eb33a
--- /dev/null
+++ b/cli/canyonos/docker_cmd.py
@@ -0,0 +1,59 @@
+"""Bounded Docker command execution for the CanyonOS CLI."""
+
+import subprocess
+import time
+
+DOCKER_QUICK_TIMEOUT = 15
+DOCKER_CLEANUP_TIMEOUT = 60
+DOCKER_RUN_TIMEOUT = 120
+DOCKER_PULL_TIMEOUT = 600
+DOCKER_CLEANUP_RETRIES = 1
+DOCKER_RETRY_DELAY = 0.5
+
+
+def run_docker(argv, *, timeout, action, check=False):
+    """Run Docker with a bounded wait and a user-facing failure message."""
+    try:
+        result = subprocess.run(
+            argv,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except FileNotFoundError:
+        raise RuntimeError("Docker is not installed or is not available on PATH.") from None
+    except subprocess.TimeoutExpired:
+        raise RuntimeError(f"{action} timed out after {timeout}s.") from None
+    except OSError as e:
+        raise RuntimeError(f"{action} could not run: {e}") from None
+
+    if check and result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        raise RuntimeError(f"{action} failed: {detail or f'exit code {result.returncode}'}")
+    return result
+
+
+def cleanup_docker(argv, *, action, missing_text=None, timeout=DOCKER_CLEANUP_TIMEOUT):
+    """Run a cleanup command, retrying once and returning its final result.
+
+    A missing resource is already clean. Other failures are returned so callers
+    can continue cleaning independent resources and report them together.
+    """
+    last_error = None
+    for attempt in range(DOCKER_CLEANUP_RETRIES + 1):
+        try:
+            result = run_docker(argv, timeout=timeout, action=action)
+        except RuntimeError as e:
+            last_error = str(e)
+        else:
+            detail = (result.stderr or result.stdout or "").strip()
+            if result.returncode == 0 or (
+                missing_text and missing_text.casefold() in detail.casefold()
+            ):
+                return None
+            last_error = f"{action} failed: {detail or f'exit code {result.returncode}'}"
+
+        if attempt < DOCKER_CLEANUP_RETRIES:
+            time.sleep(DOCKER_RETRY_DELAY)
+    return last_error
diff --git a/cli/canyonos/gc.py b/cli/canyonos/gc.py
index 44a7000e..978dc26d 100644
--- a/cli/canyonos/gc.py
+++ b/cli/canyonos/gc.py
@@ -10,8 +10,10 @@
 from canyonos import ui
 from canyonos.init import load_state
 
+REQUEST_TIMEOUT_SECONDS = 60
 
-class GCError(Exception):
+
+class GCError(RuntimeError):
     """A failed Global Controller request, carrying a message fit to print."""
 
     def __init__(self, message, code=None):
@@ -23,21 +25,33 @@ def _error_detail(e):
     """The server's `error` field, falling back to the raw body when it isn't JSON."""
     body = e.read().decode(errors="replace").strip()
     try:
-        return json.loads(body).get("error", body)
+        parsed = json.loads(body)
     except ValueError:
         return body or f"HTTP {e.code}"
+    if isinstance(parsed, dict):
+        return parsed.get("error") or body or f"HTTP {e.code}"
+    return body or f"HTTP {e.code}"
 
 
 def _request(url, action, data=None, method="GET"):
     headers = {"Content-Type": "application/json"} if data is not None else {}
     req = urllib.request.Request(url, data=data, headers=headers, method=method)
     try:
-        with urllib.request.urlopen(req) as resp:
-            return json.loads(resp.read())
+        with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_SECONDS) as resp:
+            body = resp.read()
+        try:
+            parsed = json.loads(body)
+        except (TypeError, ValueError):
+            raise GCError(f"{action} failed: Global Controller returned invalid JSON") from None
+        if not isinstance(parsed, dict):
+            raise GCError(f"{action} failed: Global Controller returned an invalid response")
+        return parsed
     except urllib.error.HTTPError as e:
         raise GCError(f"{action} failed: {_error_detail(e)}", code=e.code) from None
     except urllib.error.URLError as e:
         raise GCError(f"Could not reach Global Controller container: {e.reason}") from None
+    except (TimeoutError, OSError) as e:
+        raise GCError(f"Could not reach Global Controller container: {e}") from None
 
 
 def require_state():
@@ -55,10 +69,9 @@ def post_deploy(port, config_path=None):
     Omitting config_path lets canyonos resolve it against the synced workspace.
     """
     body = json.dumps({"config_path": config_path} if config_path else {}).encode()
-    try:
-        return _request(f"http://127.0.0.1:{port}/deploy", "Deploy", data=body, method="POST")
-    except GCError as e:
-        raise
+    return _request(
+        f"http://127.0.0.1:{port}/deploy", "Deploy", data=body, method="POST"
+    )
 
 
 def post_clean(port):
@@ -73,21 +86,35 @@ def post_clean(port):
 def workflow_endpoints(port):
     """Where the deployed workflows answer, per the container's own instance
     records -- for a workflow placed on another machine that is its public IP,
-    not this host. Empty when the container can't say (an older image has no
-    /endpoints route), which leaves the caller on its local-port fallback.
+    not this host. Empty when the container can't say, which leaves the caller
+    on its local-port fallback.
+
+    These addresses are printed, not acted on, so no failure here is fatal: an
+    older image has no /endpoints route at all, and a current one answers 200
+    with an `error` field when Redis or the config can't be read.
     """
     try:
         data = _request(f"http://127.0.0.1:{port}/endpoints", "Endpoints")
-    except GCError:
+    except GCError as e:
+        if e.code != 404:
+            ui.warn(f"Could not resolve workflow addresses: {e}")
+        return []
+    if data.get("error"):
+        ui.warn(f"Could not resolve workflow addresses: {data['error']}")
         return []
     return data.get("workflows") or []
 
 
 def deploy_status(port):
-    """Parsed /status payload, or None if the container is unreachable."""
+    """Parsed /status payload, or None when the container can't be read.
+
+    A malformed body is as uninformative as an unreachable one, so both answer
+    None rather than raising at a caller whose contract is "couldn't tell".
+    """
     url = f"http://127.0.0.1:{port}/status"
     try:
         with urllib.request.urlopen(url, timeout=5) as resp:
-            return json.loads(resp.read())
-    except OSError:
+            status = json.loads(resp.read())
+    except (OSError, ValueError, TypeError):
         return None
+    return status if isinstance(status, dict) else None
diff --git a/cli/canyonos/init.py b/cli/canyonos/init.py
index 43ded963..2bfabd2f 100644
--- a/cli/canyonos/init.py
+++ b/cli/canyonos/init.py
@@ -10,6 +10,7 @@
 import shutil
 import subprocess
 import sys
+import tempfile
 import time
 import urllib.error
 import urllib.request
@@ -18,6 +19,14 @@
 from pyfiglet import figlet_format
 
 from canyonos import ui
+from canyonos.docker_cmd import (
+    DOCKER_CLEANUP_TIMEOUT,
+    DOCKER_PULL_TIMEOUT,
+    DOCKER_QUICK_TIMEOUT,
+    DOCKER_RUN_TIMEOUT,
+    cleanup_docker,
+    run_docker,
+)
 
 
 
@@ -58,18 +67,24 @@
 
 def docker_running():
     try:
-        return subprocess.run(["docker", "info"], capture_output=True).returncode == 0
-    except OSError:
+        return run_docker(
+            ["docker", "info"],
+            timeout=DOCKER_QUICK_TIMEOUT,
+            action="docker info",
+        ).returncode == 0
+    except RuntimeError:
         return False
 
 
 def docker_start_command():
     """The command that starts the daemon for the active context, or None."""
     try:
-        result = subprocess.run(
-            ["docker", "context", "show"], capture_output=True, text=True
+        result = run_docker(
+            ["docker", "context", "show"],
+            timeout=DOCKER_QUICK_TIMEOUT,
+            action="docker context show",
         )
-    except OSError:
+    except RuntimeError:
         return None
 
     context = result.stdout.strip() if result.returncode == 0 else "default"
@@ -95,7 +110,12 @@ def ensure_docker_running(timeout=DOCKER_START_TIMEOUT):
         )
 
     ui.say(f"Docker isn't running -- starting it with `{' '.join(command)}`...")
-    subprocess.run(command, capture_output=True)
+    run_docker(
+        command,
+        timeout=DOCKER_START_TIMEOUT,
+        action="Starting the Docker runtime",
+        check=True,
+    )
 
     deadline = time.time() + timeout
     with ui.status("Waiting for the Docker daemon..."):
@@ -111,11 +131,12 @@ def ensure_docker_running(timeout=DOCKER_START_TIMEOUT):
 
 
 def pull_image(image=GC_IMAGE):
-    result = subprocess.run(["docker", "pull", image], capture_output=True, text=True)
-    if result.returncode != 0:
-        raise RuntimeError(
-            f"docker pull {image} failed: {result.stderr.strip() or result.stdout.strip()}"
-        )
+    run_docker(
+        ["docker", "pull", image],
+        timeout=DOCKER_PULL_TIMEOUT,
+        action=f"docker pull {image}",
+        check=True,
+    )
 
 
 def _port_reachable(port, attempts=10, delay=0.5):
@@ -135,11 +156,14 @@ def _port_reachable(port, attempts=10, delay=0.5):
 
 def _named_container(name=GC_CONTAINER_NAME):
     """(id, running) for the container holding exactly this name, or (None, False)."""
-    result = subprocess.run(
-        ["docker", "inspect", "--format", "{{.Id}} {{.State.Running}}", name],
-        capture_output=True,
-        text=True,
-    )
+    try:
+        result = run_docker(
+            ["docker", "inspect", "--format", "{{.Id}} {{.State.Running}}", name],
+            timeout=DOCKER_QUICK_TIMEOUT,
+            action=f"Inspecting Docker container {name}",
+        )
+    except RuntimeError:
+        return None, False
     if result.returncode != 0:
         return None, False
     fields = result.stdout.split()
@@ -168,14 +192,29 @@ def _free_container_name():
             f"Controller ({container_id[:12]}). Run `canyonos quit` to tear it down "
             "first."
         )
-    subprocess.run(["docker", "rm", "-f", container_id], capture_output=True)
+    failure = cleanup_docker(
+        ["docker", "rm", "-f", container_id],
+        action=f"Removing stale Docker container {container_id[:12]}",
+        missing_text="No such container",
+    )
+    if failure:
+        raise RuntimeError(failure)
 
 
 def run_container(image=GC_IMAGE, max_attempts=50, extra_env=None):
     port = GC_CONTAINER_PORT
     # Idempotent: succeeds silently if the network already exists (created by
     # this or a prior GC/Redis launch).
-    subprocess.run(["docker", "network", "create", LOCAL_NETWORK], capture_output=True)
+    network = run_docker(
+        ["docker", "network", "create", LOCAL_NETWORK],
+        timeout=DOCKER_QUICK_TIMEOUT,
+        action=f"Creating Docker network {LOCAL_NETWORK}",
+    )
+    if network.returncode != 0 and "already exists" not in (network.stderr or "").lower():
+        raise RuntimeError(
+            f"Could not create Docker network {LOCAL_NETWORK}: "
+            f"{network.stderr.strip() or network.stdout.strip()}"
+        )
     for _ in range(max_attempts):
         _free_container_name()
         cmd = [
@@ -204,17 +243,30 @@ def run_container(image=GC_IMAGE, max_attempts=50, extra_env=None):
         for _k, _v in (extra_env or {}).items():
             cmd.extend(["-e", f"{_k}={_v}"])
         cmd.append(image)  # image must come after all flags
-        result = subprocess.run(cmd, capture_output=True, text=True)
+        result = run_docker(
+            cmd,
+            timeout=DOCKER_RUN_TIMEOUT,
+            action="Starting the Global Controller container",
+        )
         if result.returncode == 0:
             container_id = result.stdout.strip()
+            if not container_id:
+                raise RuntimeError("Docker started the Global Controller but returned no container ID.")
             if _port_reachable(port):
                 return container_id, port
             # Port bound fine but never actually became reachable -- treat
             # like a conflict, since that's effectively what it is.
-            subprocess.run(["docker", "rm", "-f", container_id], capture_output=True)
+            failure = cleanup_docker(
+                ["docker", "rm", "-f", container_id],
+                action=f"Removing unreachable Global Controller {container_id[:12]}",
+                missing_text="No such container",
+            )
+            if failure:
+                raise RuntimeError(failure)
             port += 1
             continue
-        if "port is already allocated" in result.stderr:
+        error = (result.stderr or "").lower()
+        if "port is already allocated" in error or "address already in use" in error:
             port += 1
             continue
         raise RuntimeError(result.stderr)
@@ -222,16 +274,52 @@ def run_container(image=GC_IMAGE, max_attempts=50, extra_env=None):
 
 
 def save_state(container_id, port):
-    """ Writes GC container info to ~/.canyonos/state.json"""
+    """Atomically write GC state so interruption cannot truncate the old record."""
     os.makedirs(STATE_DIR, exist_ok=True)
-    with open(STATE_PATH, "w") as f:
-        json.dump({"container_id": container_id, "port": port}, f)
+    temporary_path = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", dir=STATE_DIR, prefix="state.", suffix=".tmp", delete=False
+        ) as f:
+            temporary_path = f.name
+            json.dump({"container_id": container_id, "port": port}, f)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(temporary_path, STATE_PATH)
+    except Exception:
+        if temporary_path is not None:
+            try:
+                os.remove(temporary_path)
+            except FileNotFoundError:
+                pass
+        raise
 
 
 def load_state():
     """Reads GC container info from ~/.canyonos/state.json"""
-    with open(STATE_PATH) as f:
-        return json.load(f)
+    try:
+        with open(STATE_PATH) as f:
+            state = json.load(f)
+    except FileNotFoundError:
+        raise
+    except (OSError, json.JSONDecodeError) as e:
+        raise RuntimeError(
+            f"Could not read CanyonOS state at {STATE_PATH}: {e}. "
+            "Move or remove that file, then run `canyonos deploy` again."
+        ) from None
+
+    if (
+        not isinstance(state, dict)
+        or not isinstance(state.get("container_id"), str)
+        or not state["container_id"]
+        or not isinstance(state.get("port"), int)
+        or not 1 <= state["port"] <= 65535
+    ):
+        raise RuntimeError(
+            f"CanyonOS state at {STATE_PATH} is invalid. "
+            "Move or remove that file, then run `canyonos deploy` again."
+        )
+    return state
 
 
 def quit_existing():
@@ -243,8 +331,8 @@ def quit_existing():
     # Deferred: quit.py imports from this module, so a top-level import cycles.
     from canyonos.quit import run_quit
 
-    if os.path.isfile(STATE_PATH):
-        run_quit()
+    if os.path.isfile(STATE_PATH) and run_quit():
+        ui.warn("Starting a new Global Controller anyway; the leftovers above need cleaning up by hand.")
 
 
 def run_init(banner=True, extra_env=None):
@@ -259,5 +347,17 @@ def run_init(banner=True, extra_env=None):
         pull_image()
     with ui.status("Starting Global Controller container..."):
         container_id, port = run_container(extra_env=extra_env)
-    save_state(container_id, port)
+    try:
+        save_state(container_id, port)
+    except OSError as e:
+        cleanup_failure = cleanup_docker(
+            ["docker", "rm", "-f", container_id],
+            action=f"Removing Global Controller {container_id[:12]}",
+            missing_text="No such container",
+        )
+        detail = f" Automatic cleanup also failed: {cleanup_failure}" if cleanup_failure else ""
+        raise RuntimeError(
+            f"Could not save Global Controller state at {STATE_PATH}: {e}. "
+            f"Removed container {container_id[:12]}.{detail}"
+        ) from None
     ui.ok(f"Global Controller running in container {container_id[:12]} on port {port}")
diff --git a/cli/canyonos/logs.py b/cli/canyonos/logs.py
index 3969af62..efa71d1c 100644
--- a/cli/canyonos/logs.py
+++ b/cli/canyonos/logs.py
@@ -11,19 +11,23 @@
 def run_logs():
     state = require_state()
     if state is None:
-        return
+        raise RuntimeError("No Global Controller container is available for logs.")
 
     status = deploy_status(state["port"])
     if status is None:
-        ui.fail("Could not reach Global Controller container.")
-        return
+        raise RuntimeError("Could not reach Global Controller container.")
 
     if not status.get("running"):
-        ui.warn("No deploy running, run `canyonos deploy` to deploy project.")
-        return
+        raise RuntimeError("No deploy is running. Run `canyonos deploy` first.")
 
     try:
-        subprocess.run(["docker", "logs", "-f", state["container_id"]])
+        result = subprocess.run(["docker", "logs", "-f", state["container_id"]])
     except KeyboardInterrupt:
         ui.blank()
         ui.say("Stopped monitoring log stream. Run `canyonos stop` to stop the deploy.")
+        return
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"Docker log stream failed with exit code {result.returncode}. "
+            "Run `canyonos status` to check the deploy."
+        )
diff --git a/cli/canyonos/quit.py b/cli/canyonos/quit.py
index f7ee16c6..23f4e97f 100644
--- a/cli/canyonos/quit.py
+++ b/cli/canyonos/quit.py
@@ -7,54 +7,116 @@
 """
 
 import os
-import subprocess
 
 from canyonos import ui
+from canyonos.docker_cmd import (
+    DOCKER_CLEANUP_TIMEOUT,
+    DOCKER_QUICK_TIMEOUT,
+    cleanup_docker,
+    run_docker,
+)
 from canyonos.dashboard_stack import teardown_dashboard
 from canyonos.gc import GCError, post_clean, require_state
 from canyonos.init import GC_WORKSPACE_VOLUME, STATE_PATH
 
 
-def _container_exists(container_id):
-    result = subprocess.run(
-        ["docker", "inspect", container_id], capture_output=True
-    )
-    return result.returncode == 0
+def _container_state(container_id):
+    """(exists, running), as far as Docker will say.
+
+    An inspect Docker can't answer at all reads as a container that is still
+    there and still up, so teardown attempts every removal instead of skipping
+    to deleting the state that records what to remove.
+    """
+    try:
+        result = run_docker(
+            ["docker", "inspect", "--format", "{{.State.Running}}", container_id],
+            timeout=DOCKER_QUICK_TIMEOUT,
+            action=f"Inspecting Docker container {container_id[:12]}",
+        )
+    except RuntimeError:
+        return True, True
+    if result.returncode != 0:
+        return False, False
+    return True, result.stdout.strip() == "true"
 
 
 def run_quit():
+    """Tear down the container, volume and dashboard; returns what it couldn't.
+
+    Teardown reports rather than raises. It runs both on its own and as the
+    first step of `canyonos deploy`, so a resource that won't go away must not
+    also block starting over -- the state file is preserved either way, which
+    is what makes a later retry possible.
+    """
     state = require_state()
     if state is None:
-        return
+        return []
 
     container_id = state["container_id"]
     with ui.status("Tearing down..."):
+        failures = []
+        # A stale record cannot answer /clean, so establish this before making
+        # an HTTP request that would otherwise add a full request timeout.
+        exists, container_running = _container_state(container_id)
+
         # Stop any running deploy first, so the local controller and Redis
         # containers it spawned via docker-outside-of-docker get torn down
         # too. Removing the GC container itself doesn't touch them -- they're
         # sibling containers on the host, not nested inside it.
-        try:
-            post_clean(state["port"])
-        except GCError:
-            # Nothing was running, or the GC is already unreachable/gone.
-            pass
-
-        # state.json can go stale (daemon restarted, container removed by
-        # hand, a previous `quit` died partway through) -- don't let a
-        # missing container turn `quit` into a crash instead of a cleanup.
-        already_gone = not _container_exists(container_id)
-        if not already_gone:
-            subprocess.run(["docker", "stop", container_id], check=False, capture_output=True)
-            subprocess.run(["docker", "rm", container_id], check=False, capture_output=True)
+        if container_running:
+            try:
+                post_clean(state["port"])
+            except GCError as e:
+                # 409 is "no deploy running", which is the desired end state.
+                if e.code != 409:
+                    failures.append(f"Cleaning the running deployment failed: {e}")
+
+            failure = cleanup_docker(
+                ["docker", "stop", container_id],
+                action=f"Stopping Global Controller {container_id[:12]}",
+                missing_text="No such container",
+                timeout=DOCKER_CLEANUP_TIMEOUT,
+            )
+            if failure:
+                failures.append(failure)
+
+        if exists:
+            failure = cleanup_docker(
+                ["docker", "rm", container_id],
+                action=f"Removing Global Controller {container_id[:12]}",
+                missing_text="No such container",
+                timeout=DOCKER_CLEANUP_TIMEOUT,
+            )
+            if failure:
+                failures.append(failure)
 
         # Remove the workspace volume only after the container is gone (docker
-        # refuses to remove a volume still in use). check=False so a missing
-        # volume doesn't turn teardown into an error.
-        subprocess.run(["docker", "volume", "rm", GC_WORKSPACE_VOLUME], check=False, capture_output=True)
-        teardown_dashboard()
-        os.remove(STATE_PATH)
+        # refuses to remove a volume still in use). A missing volume is already
+        # the desired end state; every other failure remains retryable.
+        failure = cleanup_docker(
+            ["docker", "volume", "rm", GC_WORKSPACE_VOLUME],
+            action=f"Removing workspace volume {GC_WORKSPACE_VOLUME}",
+            missing_text="No such volume",
+            timeout=DOCKER_CLEANUP_TIMEOUT,
+        )
+        if failure:
+            failures.append(failure)
+        if not teardown_dashboard():
+            failures.append("Removing the dashboard stack failed")
+
+        if not failures:
+            try:
+                os.remove(STATE_PATH)
+            except FileNotFoundError:
+                pass
+            except OSError as e:
+                failures.append(f"Removing CanyonOS state at {STATE_PATH} failed: {e}")
 
-    if already_gone:
+    if failures:
+        ui.fail("Teardown incomplete:\n- " + "\n- ".join(failures))
+        ui.hint(f"State was preserved at {STATE_PATH}, so `canyonos quit` can be retried.")
+    elif not exists:
         ui.warn(f"Global Controller container {container_id[:12]} was already gone; cleaned up local state.")
     else:
         ui.ok(f"Global Controller container {container_id[:12]} torn down (volume removed)")
+    return failures
diff --git a/cli/canyonos/stop.py b/cli/canyonos/stop.py
index 204c3b2f..e7012861 100644
--- a/cli/canyonos/stop.py
+++ b/cli/canyonos/stop.py
@@ -16,7 +16,12 @@ def run_stop():
     try:
         with ui.status("Stopping deploy..."):
             post_clean(state["port"])
-            stop_dashboard()
-        ui.ok("Deploy stopped.")
+            dashboard_stopped = stop_dashboard()
     except GCError as e:
         ui.fail(e)
+        return
+
+    if dashboard_stopped:
+        ui.ok("Deploy stopped.")
+    else:
+        ui.warn("Deploy stopped, but the dashboard did not. Run `canyonos quit` to remove it.")
diff --git a/cli/canyonos/sync.py b/cli/canyonos/sync.py
index 84f875cd..ab8162de 100644
--- a/cli/canyonos/sync.py
+++ b/cli/canyonos/sync.py
@@ -1,25 +1,226 @@
 """
-Logic for `canyonos sync`: copy the current project directory into the Global
+Logic for `canyonos sync`: mirror the current project directory into the Global
 Controller container's /workspace volume via `docker cp`.
 
 Files live inside the container's named volume (see `init.py`), not on a live
-bind mount, so host-side edits don't reach a running build. `docker cp` is
-additive -- it overwrites and adds but never deletes -- so a standalone
-re-sync leaves behind anything removed from the host since the last one.
-That can't accumulate across deploys: `canyonos deploy` quits any previous
-controller first, which removes the volume.
+bind mount, so host-side edits don't reach a running build. A manifest records
+which paths the previous sync managed. Before the next additive `docker cp`,
+only managed paths that disappeared from the host are removed; container-side
+build artifacts and other untracked workspace files are left alone.
 """
 
+import json
 import os
-import subprocess
+import tempfile
 
 from canyonos import ui
+from canyonos.docker_cmd import DOCKER_RUN_TIMEOUT, run_docker
 from canyonos.gc import require_state
-from canyonos.init import GC_WORKSPACE_PATH
+from canyonos.init import GC_WORKSPACE_PATH, STATE_DIR
+
+
+SYNC_STATE_PATH = os.path.join(STATE_DIR, "sync-manifest.json")
+_SYNC_STATE_VERSION = 1
+
+# This runs inside the Global Controller container. Paths are deleted deepest
+# first, and directories are removed only when empty. That preserves files a
+# container-side build may have generated beneath a formerly synced directory.
+_DELETE_STALE_SCRIPT = """
+import errno
+import json
+import os
+import sys
+
+root = os.path.realpath(sys.argv[1])
+manifest_path = sys.argv[2]
+try:
+    with open(manifest_path, encoding="utf-8") as manifest_file:
+        stale_entries = json.load(manifest_file)
+    for relative, kind in stale_entries:
+        parts = relative.split("/")
+        if not relative or any(part in ("", ".", "..") for part in parts):
+            raise RuntimeError("unsafe sync manifest path: " + repr(relative))
+        unresolved_parent = os.path.join(root, *parts[:-1])
+        parent = os.path.realpath(unresolved_parent)
+        if parent != os.path.abspath(unresolved_parent):
+            raise RuntimeError("sync manifest path crosses a symlink: " + repr(relative))
+        if os.path.commonpath((root, parent)) != root:
+            raise RuntimeError("sync manifest path escapes workspace: " + repr(relative))
+        target = os.path.join(parent, parts[-1])
+        if kind == "directory":
+            try:
+                os.rmdir(target)
+            except FileNotFoundError:
+                pass
+            except OSError as error:
+                if error.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOTDIR):
+                    raise
+        else:
+            try:
+                if not os.path.isdir(target) or os.path.islink(target):
+                    os.unlink(target)
+            except FileNotFoundError:
+                pass
+finally:
+    try:
+        os.unlink(manifest_path)
+    except FileNotFoundError:
+        pass
+"""
+
+
+def _project_entries(project_root):
+    """Return the relative paths and types managed by a project sync."""
+    entries = {}
+    for directory, dirnames, filenames in os.walk(project_root, followlinks=False):
+        for name in dirnames + filenames:
+            path = os.path.join(directory, name)
+            relative = os.path.relpath(path, project_root).replace(os.sep, "/")
+            entries[relative] = (
+                "directory"
+                if os.path.isdir(path) and not os.path.islink(path)
+                else "entry"
+            )
+    return entries
+
+
+# The manifest only records what the last sync copied, so discarding it costs
+# nothing beyond skipping stale-file removal on the next run.
+_MANIFEST_RECOVERY_HINT = (
+    "Move or remove that file, then run `canyonos deploy` again."
+)
+
+
+def _previous_entries(container_id):
+    try:
+        with open(SYNC_STATE_PATH, encoding="utf-8") as manifest_file:
+            manifest = json.load(manifest_file)
+    except FileNotFoundError:
+        return {}
+    except (OSError, json.JSONDecodeError) as error:
+        raise RuntimeError(
+            f"Could not read the sync manifest at {SYNC_STATE_PATH}: {error}. "
+            f"{_MANIFEST_RECOVERY_HINT}"
+        ) from None
+
+    if not isinstance(manifest, dict):
+        raise RuntimeError(
+            f"The sync manifest at {SYNC_STATE_PATH} is invalid. {_MANIFEST_RECOVERY_HINT}"
+        )
+
+    if manifest.get("container_id") != container_id:
+        return {}
+
+    entries = manifest.get("entries")
+    if (
+        manifest.get("version") != _SYNC_STATE_VERSION
+        or not isinstance(entries, dict)
+        or any(
+            not isinstance(path, str)
+            or not path
+            or path.startswith("/")
+            or any(part in ("", ".", "..") for part in path.split("/"))
+            or kind not in {"directory", "entry"}
+            for path, kind in entries.items()
+        )
+    ):
+        raise RuntimeError(
+            f"The sync manifest at {SYNC_STATE_PATH} is invalid. {_MANIFEST_RECOVERY_HINT}"
+        )
+    return entries
+
+
+def _stale_entries(previous, current):
+    stale = [
+        (path, kind)
+        for path, kind in previous.items()
+        if current.get(path) != kind
+    ]
+    return sorted(stale, key=lambda item: item[0].count("/"), reverse=True)
+
+
+def _remove_stale_entries(container_id, stale_entries):
+    if not stale_entries:
+        return
+
+    local_path = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", prefix="canyonos-sync-stale-", suffix=".json", delete=False
+        ) as manifest_file:
+            local_path = manifest_file.name
+            json.dump(stale_entries, manifest_file)
+
+        remote_path = f"/tmp/{os.path.basename(local_path)}"
+        copied = run_docker(
+            ["docker", "cp", local_path, f"{container_id}:{remote_path}"],
+            timeout=DOCKER_RUN_TIMEOUT,
+            action="Preparing stale project file cleanup",
+        )
+        if copied.returncode != 0:
+            detail = copied.stderr.strip() or copied.stdout.strip()
+            raise RuntimeError(f"Could not prepare stale project file cleanup: {detail}")
+
+        removed = run_docker(
+            [
+                "docker",
+                "exec",
+                container_id,
+                "python",
+                "-c",
+                _DELETE_STALE_SCRIPT,
+                GC_WORKSPACE_PATH,
+                remote_path,
+            ],
+            timeout=DOCKER_RUN_TIMEOUT,
+            action="Removing stale project files",
+        )
+        if removed.returncode != 0:
+            detail = removed.stderr.strip() or removed.stdout.strip()
+            raise RuntimeError(f"Could not remove stale project files: {detail}")
+    finally:
+        if local_path is not None:
+            try:
+                os.remove(local_path)
+            except FileNotFoundError:
+                pass
+
+
+def _save_sync_state(container_id, entries):
+    os.makedirs(os.path.dirname(SYNC_STATE_PATH), exist_ok=True)
+    temporary_path = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            dir=os.path.dirname(SYNC_STATE_PATH),
+            prefix="sync-manifest.",
+            suffix=".tmp",
+            delete=False,
+        ) as manifest_file:
+            temporary_path = manifest_file.name
+            json.dump(
+                {
+                    "version": _SYNC_STATE_VERSION,
+                    "container_id": container_id,
+                    "entries": entries,
+                },
+                manifest_file,
+                sort_keys=True,
+            )
+            manifest_file.flush()
+            os.fsync(manifest_file.fileno())
+        os.replace(temporary_path, SYNC_STATE_PATH)
+    except Exception:
+        if temporary_path is not None:
+            try:
+                os.remove(temporary_path)
+            except FileNotFoundError:
+                pass
+        raise
 
 
 def run_sync():
-    """Copy the current directory into the container. Returns True on success."""
+    """Mirror the current directory into the container. Returns True on success."""
     state = require_state()
     if state is None:
         return False
@@ -32,14 +233,29 @@ def run_sync():
 
     with ui.status(f"{label}..."):
         # Captured so docker's own progress output doesn't clobber the spinner.
-        result = subprocess.run(
-            ["docker", "cp", src, f"{container_id}:{GC_WORKSPACE_PATH}"],
-            capture_output=True,
-            text=True,
-        )
+        try:
+            entries = _project_entries(os.getcwd())
+            stale_entries = _stale_entries(
+                _previous_entries(container_id), entries
+            )
+            _remove_stale_entries(container_id, stale_entries)
+            result = run_docker(
+                ["docker", "cp", src, f"{container_id}:{GC_WORKSPACE_PATH}"],
+                timeout=DOCKER_RUN_TIMEOUT,
+                action="Syncing project files",
+            )
+        except (OSError, RuntimeError) as e:
+            ui.fail(str(e))
+            return False
     if result.returncode != 0:
         ui.fail(f"Sync failed: {result.stderr.strip() or result.stdout.strip()}")
         return False
 
+    try:
+        _save_sync_state(container_id, entries)
+    except OSError as e:
+        ui.fail(f"Could not save the sync manifest at {SYNC_STATE_PATH}: {e}")
+        return False
+
     ui.ok("Sync complete.")
     return True
diff --git a/cli/cli.py b/cli/cli.py
index 2d38c612..b5140b59 100644
--- a/cli/cli.py
+++ b/cli/cli.py
@@ -39,6 +39,12 @@ def print_help(self, file=None):
 
 def main():
     parser = _RootParser(prog="canyonos")
+    parser.add_argument(
+        "-v",
+        "--version",
+        action="version",
+        version=f"canyonos {importlib.metadata.version('canyonos')}",
+    )
     # Subparsers keep the stock argparse help, so `canyonos <cmd> -h` still
     # describes that command instead of reprinting the top-level screen.
     subparsers = parser.add_subparsers(dest="command", parser_class=argparse.ArgumentParser)
@@ -79,7 +85,6 @@ def add(name, run):
     add("config", lambda args: run_config())
     add("build", lambda args: run_build())
     add("doctor", lambda args: sys.exit(0 if run_doctor() else 1))
-    add("version", lambda args: ui.say(f"canyonos {importlib.metadata.version('canyonos')}"))
     add("serve", lambda args: sys.exit(run_serve()))
     add("status", lambda args: run_status())
 
diff --git a/cli/utils/help_screen.py b/cli/utils/help_screen.py
index 717e021c..61d58351 100644
--- a/cli/utils/help_screen.py
+++ b/cli/utils/help_screen.py
@@ -29,7 +29,6 @@
     ("status", "Check whether a deploy is running and where it answers"),
     ("stop", "Stop the running deploy, keeping the container and files"),
     ("test", "Deploy locally and run one prompt end to end"),
-    ("version", "Print canyonos version"),
 )
 
 DESCRIPTIONS = dict(CORE_COMMANDS + UTIL_COMMANDS)
diff --git a/images/canyonos-banner.gif b/images/canyonos-banner.gif
index 0a110876..15b35916 100644
Binary files a/images/canyonos-banner.gif and b/images/canyonos-banner.gif differ
diff --git a/images/financial_analyst_results_page.jpg b/images/financial_analyst_results_page.jpg
deleted file mode 100644
index 9e59c76e..00000000
Binary files a/images/financial_analyst_results_page.jpg and /dev/null differ
diff --git a/images/ventis-logo.png b/images/ventis-logo.png
deleted file mode 100644
index b26b8606..00000000
Binary files a/images/ventis-logo.png and /dev/null differ
diff --git a/tests/test_canyonos_deploy.py b/tests/test_canyonos_deploy.py
index ac62c88d..216acf5a 100644
--- a/tests/test_canyonos_deploy.py
+++ b/tests/test_canyonos_deploy.py
@@ -17,6 +17,8 @@ def deployable(monkeypatch):
     monkeypatch.setattr(deploy_cmd, "workflow_api_port", lambda _config: 8080)
     monkeypatch.setattr(deploy_cmd, "port_in_use", lambda _port: False)
     monkeypatch.setattr(deploy_cmd, "post_deploy", lambda *_a: None)
+    monkeypatch.setattr(deploy_cmd, "_start_dashboard", lambda: None)
+    monkeypatch.setattr(deploy_cmd, "run_quit", lambda: None)
 
 
 def test_a_config_path_outside_the_project_raises(monkeypatch, deployable):
@@ -27,11 +29,30 @@ def test_a_config_path_outside_the_project_raises(monkeypatch, deployable):
 
 
 def test_a_sync_failure_raises(monkeypatch, deployable):
+    cleaned = []
     monkeypatch.setattr(deploy_cmd, "run_sync", lambda: False)
+    monkeypatch.setattr(deploy_cmd, "run_quit", lambda: cleaned.append(True))
 
     with pytest.raises(RuntimeError, match="Could not sync the project"):
         deploy_cmd.run_deploy(CONFIG_PATH, quiet=True)
 
+    assert cleaned == [True]
+
+
+def test_ctrl_c_during_setup_cleans_up_the_partial_deploy(monkeypatch, deployable):
+    cleaned = []
+    monkeypatch.setattr(
+        deploy_cmd,
+        "run_sync",
+        lambda: (_ for _ in ()).throw(KeyboardInterrupt()),
+    )
+    monkeypatch.setattr(deploy_cmd, "run_quit", lambda: cleaned.append(True))
+
+    with pytest.raises(KeyboardInterrupt):
+        deploy_cmd.run_deploy(CONFIG_PATH, quiet=True)
+
+    assert cleaned == [True]
+
 
 def test_an_occupied_api_port_raises_before_post_deploy(monkeypatch, deployable):
     """Checked after run_init() (which already tore down any previous deploy), so a still-live
@@ -46,6 +67,21 @@ def test_an_occupied_api_port_raises_before_post_deploy(monkeypatch, deployable)
     assert calls == []
 
 
+def test_a_log_monitor_failure_cleans_up_the_partial_deploy(monkeypatch, deployable):
+    cleaned = []
+    monkeypatch.setattr(
+        deploy_cmd,
+        "_stream_logs_and_autoserve",
+        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("logs broke")),
+    )
+    monkeypatch.setattr(deploy_cmd, "run_quit", lambda: cleaned.append(True))
+
+    with pytest.raises(RuntimeError, match="logs broke"):
+        deploy_cmd.run_deploy(CONFIG_PATH, serve=False)
+
+    assert cleaned == [True]
+
+
 def test_a_post_deploy_failure_is_reraised_as_a_runtime_error(monkeypatch, deployable):
     def boom(*_a):
         raise GCError("Deploy failed: conflict")
@@ -56,6 +92,24 @@ def boom(*_a):
         deploy_cmd.run_deploy(CONFIG_PATH, quiet=True)
 
 
+def test_a_failure_after_the_workflow_is_up_leaves_the_deploy_alone(monkeypatch, deployable):
+    """Summary rendering runs after the workflow is serving, so anything that
+    fails there is a reporting problem -- not a reason to tear the stack down."""
+    cleaned = []
+
+    def ready_then_fail(*_a, on_ready, **_k):
+        on_ready(True)
+        raise RuntimeError("could not resolve endpoints")
+
+    monkeypatch.setattr(deploy_cmd, "_stream_logs_and_autoserve", ready_then_fail)
+    monkeypatch.setattr(deploy_cmd, "run_quit", lambda: cleaned.append(True))
+
+    with pytest.raises(RuntimeError, match="could not resolve endpoints"):
+        deploy_cmd.run_deploy(CONFIG_PATH, serve=False)
+
+    assert cleaned == []
+
+
 def test_quiet_returns_state_without_streaming(monkeypatch, deployable):
     def unexpected(*_a, **_k):
         raise AssertionError("quiet=True should skip the log-tail/dashboard UI")
@@ -69,7 +123,7 @@ def test_non_quiet_still_streams_and_returns_state(monkeypatch, deployable):
     calls = []
     monkeypatch.setattr(
         deploy_cmd, "_stream_logs_and_autoserve",
-        lambda state, api_port, config_path, serve, verbose: calls.append(
+        lambda state, api_port, config_path, serve, verbose, on_ready: calls.append(
             (state, api_port, config_path, serve, verbose)
         ),
     )
diff --git a/tests/test_canyonos_init_hardening.py b/tests/test_canyonos_init_hardening.py
new file mode 100644
index 00000000..1f2e10fa
--- /dev/null
+++ b/tests/test_canyonos_init_hardening.py
@@ -0,0 +1,95 @@
+import json
+import subprocess
+
+import pytest
+
+from canyonos import init as init_cmd
+
+
+def test_docker_runtime_start_uses_bounded_translated_command(monkeypatch):
+    calls = []
+    monkeypatch.setattr(init_cmd, "docker_running", lambda: False)
+    monkeypatch.setattr(init_cmd, "docker_start_command", lambda: ["orb", "start"])
+
+    def failed(argv, **kwargs):
+        calls.append((argv, kwargs))
+        raise RuntimeError("Starting the Docker runtime failed: launch failed")
+
+    monkeypatch.setattr(init_cmd, "run_docker", failed)
+
+    with pytest.raises(RuntimeError, match="Starting the Docker runtime failed: launch failed"):
+        init_cmd.ensure_docker_running()
+
+    assert calls == [
+        (
+            ["orb", "start"],
+            {
+                "timeout": init_cmd.DOCKER_START_TIMEOUT,
+                "action": "Starting the Docker runtime",
+                "check": True,
+            },
+        )
+    ]
+
+
+def test_failed_state_write_preserves_the_previous_state(monkeypatch, tmp_path):
+    state_path = tmp_path / "state.json"
+    original = {"container_id": "old", "port": 8000}
+    state_path.write_text(json.dumps(original))
+    monkeypatch.setattr(init_cmd, "STATE_DIR", str(tmp_path))
+    monkeypatch.setattr(init_cmd, "STATE_PATH", str(state_path))
+    monkeypatch.setattr(
+        init_cmd.json,
+        "dump",
+        lambda *_a, **_k: (_ for _ in ()).throw(OSError("disk full")),
+    )
+
+    with pytest.raises(OSError, match="disk full"):
+        init_cmd.save_state("new", 8001)
+
+    assert json.loads(state_path.read_text()) == original
+    assert list(tmp_path.glob("state.*.tmp")) == []
+
+
+def test_state_save_failure_removes_the_new_container(monkeypatch):
+    calls = []
+    monkeypatch.setattr(init_cmd, "ensure_docker_running", lambda: None)
+    monkeypatch.setattr(init_cmd, "quit_existing", lambda: None)
+    monkeypatch.setattr(init_cmd, "pull_image", lambda: None)
+    monkeypatch.setattr(
+        init_cmd, "run_container", lambda extra_env=None: ("abcdef123456", 8000)
+    )
+    monkeypatch.setattr(
+        init_cmd, "save_state", lambda *_a: (_ for _ in ()).throw(OSError("disk full"))
+    )
+    monkeypatch.setattr(
+        init_cmd.subprocess,
+        "run",
+        lambda argv, **_k: (
+            calls.append(argv) or subprocess.CompletedProcess(argv, 0, "", "")
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="disk full"):
+        init_cmd.run_init(banner=False)
+
+    assert calls == [["docker", "rm", "-f", "abcdef123456"]]
+
+
+@pytest.mark.parametrize(
+    "contents",
+    [
+        "",
+        "not json",
+        "[]",
+        '{"container_id": "abc"}',
+        '{"container_id": "abc", "port": 70000}',
+    ],
+)
+def test_invalid_state_has_recovery_guidance(monkeypatch, tmp_path, contents):
+    state_path = tmp_path / "state.json"
+    state_path.write_text(contents)
+    monkeypatch.setattr(init_cmd, "STATE_PATH", str(state_path))
+
+    with pytest.raises(RuntimeError, match="Move or remove that file"):
+        init_cmd.load_state()
diff --git a/tests/test_canyonos_logs.py b/tests/test_canyonos_logs.py
new file mode 100644
index 00000000..9b4a5778
--- /dev/null
+++ b/tests/test_canyonos_logs.py
@@ -0,0 +1,30 @@
+import subprocess
+
+import pytest
+
+from canyonos import logs
+
+
+def test_unreachable_controller_fails_logs(monkeypatch):
+    monkeypatch.setattr(
+        logs, "require_state", lambda: {"container_id": "abc", "port": 8000}
+    )
+    monkeypatch.setattr(logs, "deploy_status", lambda _port: None)
+
+    with pytest.raises(RuntimeError, match="Could not reach"):
+        logs.run_logs()
+
+
+def test_failed_docker_log_stream_is_not_reported_as_success(monkeypatch):
+    monkeypatch.setattr(
+        logs, "require_state", lambda: {"container_id": "abc", "port": 8000}
+    )
+    monkeypatch.setattr(logs, "deploy_status", lambda _port: {"running": True})
+    monkeypatch.setattr(
+        logs.subprocess,
+        "run",
+        lambda argv: subprocess.CompletedProcess(argv, 17),
+    )
+
+    with pytest.raises(RuntimeError, match="exit code 17"):
+        logs.run_logs()
diff --git a/tests/test_canyonos_quit_hardening.py b/tests/test_canyonos_quit_hardening.py
new file mode 100644
index 00000000..a6950dc9
--- /dev/null
+++ b/tests/test_canyonos_quit_hardening.py
@@ -0,0 +1,112 @@
+import subprocess
+
+from canyonos import docker_cmd, quit as quit_cmd
+from canyonos.gc import GCError
+
+
+def _state(monkeypatch, tmp_path):
+    state_path = tmp_path / "state.json"
+    state_path.write_text('{"container_id": "abcdef123456", "port": 8000}')
+    monkeypatch.setattr(quit_cmd, "STATE_PATH", str(state_path))
+    monkeypatch.setattr(
+        quit_cmd,
+        "require_state",
+        lambda: {"container_id": "abcdef123456", "port": 8000},
+    )
+    monkeypatch.setattr(quit_cmd, "_container_state", lambda _container_id: (True, True))
+    monkeypatch.setattr(quit_cmd, "teardown_dashboard", lambda: True)
+    return state_path
+
+
+def test_controller_cleanup_failure_preserves_state(monkeypatch, tmp_path):
+    state_path = _state(monkeypatch, tmp_path)
+    docker_calls = []
+    monkeypatch.setattr(
+        quit_cmd,
+        "post_clean",
+        lambda _port: (_ for _ in ()).throw(GCError("controller stuck", code=500)),
+    )
+
+    def docker(argv, **_kwargs):
+        docker_calls.append(argv)
+        return subprocess.CompletedProcess(argv, 0, "", "")
+
+    monkeypatch.setattr(docker_cmd.subprocess, "run", docker)
+
+    failures = quit_cmd.run_quit()
+
+    assert any("controller stuck" in failure for failure in failures)
+    assert state_path.exists()
+    # A controller that won't clean doesn't stop the rest of the teardown.
+    assert ["docker", "rm", "abcdef123456"] in docker_calls
+
+
+def test_docker_cleanup_failure_preserves_state(monkeypatch, tmp_path):
+    state_path = _state(monkeypatch, tmp_path)
+    monkeypatch.setattr(quit_cmd, "post_clean", lambda _port: None)
+
+    def docker(argv, **_kwargs):
+        return subprocess.CompletedProcess(
+            argv,
+            1 if argv[:2] == ["docker", "rm"] else 0,
+            "",
+            "daemon error" if argv[:2] == ["docker", "rm"] else "",
+        )
+
+    monkeypatch.setattr(docker_cmd.subprocess, "run", docker)
+
+    failures = quit_cmd.run_quit()
+
+    assert any("daemon error" in failure for failure in failures)
+    assert state_path.exists()
+
+
+def test_successful_cleanup_removes_state(monkeypatch, tmp_path):
+    state_path = _state(monkeypatch, tmp_path)
+    monkeypatch.setattr(quit_cmd, "post_clean", lambda _port: None)
+    monkeypatch.setattr(
+        docker_cmd.subprocess,
+        "run",
+        lambda argv, **_kwargs: subprocess.CompletedProcess(argv, 0, "", ""),
+    )
+
+    assert quit_cmd.run_quit() == []
+
+    assert not state_path.exists()
+
+
+def test_stopped_controller_is_removed_without_calling_clean(monkeypatch, tmp_path):
+    state_path = _state(monkeypatch, tmp_path)
+    monkeypatch.setattr(quit_cmd, "_container_state", lambda _container_id: (True, False))
+    clean_calls = []
+    docker_calls = []
+    monkeypatch.setattr(quit_cmd, "post_clean", lambda _port: clean_calls.append(True))
+
+    def docker(argv, **_kwargs):
+        docker_calls.append(argv)
+        return subprocess.CompletedProcess(argv, 0, "", "")
+
+    monkeypatch.setattr(docker_cmd.subprocess, "run", docker)
+
+    assert quit_cmd.run_quit() == []
+
+    assert clean_calls == []
+    assert ["docker", "stop", "abcdef123456"] not in docker_calls
+    assert ["docker", "rm", "abcdef123456"] in docker_calls
+    assert not state_path.exists()
+
+
+def test_dashboard_teardown_failure_preserves_state(monkeypatch, tmp_path):
+    state_path = _state(monkeypatch, tmp_path)
+    monkeypatch.setattr(quit_cmd, "post_clean", lambda _port: None)
+    monkeypatch.setattr(quit_cmd, "teardown_dashboard", lambda: False)
+    monkeypatch.setattr(
+        docker_cmd.subprocess,
+        "run",
+        lambda argv, **_kwargs: subprocess.CompletedProcess(argv, 0, "", ""),
+    )
+
+    failures = quit_cmd.run_quit()
+
+    assert any("dashboard stack failed" in failure for failure in failures)
+    assert state_path.exists()
diff --git a/tests/test_canyonos_stop_hardening.py b/tests/test_canyonos_stop_hardening.py
new file mode 100644
index 00000000..1d63b34e
--- /dev/null
+++ b/tests/test_canyonos_stop_hardening.py
@@ -0,0 +1,18 @@
+from canyonos import stop as stop_cmd
+
+
+def test_dashboard_stop_failure_is_not_reported_as_success(monkeypatch):
+    reported = []
+    monkeypatch.setattr(
+        stop_cmd,
+        "require_state",
+        lambda: {"container_id": "abcdef123456", "port": 8000},
+    )
+    monkeypatch.setattr(stop_cmd, "post_clean", lambda _port: None)
+    monkeypatch.setattr(stop_cmd, "stop_dashboard", lambda: False)
+    monkeypatch.setattr(stop_cmd.ui, "ok", lambda _message: reported.append("ok"))
+    monkeypatch.setattr(stop_cmd.ui, "warn", lambda message: reported.append(str(message)))
+
+    stop_cmd.run_stop()
+
+    assert reported == ["Deploy stopped, but the dashboard did not. Run `canyonos quit` to remove it."]
diff --git a/tests/test_canyonos_sync.py b/tests/test_canyonos_sync.py
new file mode 100644
index 00000000..fd4461da
--- /dev/null
+++ b/tests/test_canyonos_sync.py
@@ -0,0 +1,68 @@
+import contextlib
+import shutil
+import subprocess
+import sys
+
+from canyonos import sync as sync_cmd
+
+
+def test_resync_removes_a_host_file_deleted_since_the_previous_sync(
+    monkeypatch, tmp_path
+):
+    source = tmp_path / "source"
+    destination = tmp_path / "workspace"
+    source.mkdir()
+    destination.mkdir()
+    (source / "a.txt").write_text("A")
+    (source / "b.txt").write_text("B")
+    (source / "managed").mkdir()
+    (source / "managed" / "source.txt").write_text("managed")
+
+    monkeypatch.chdir(source)
+    monkeypatch.setattr(
+        sync_cmd,
+        "require_state",
+        lambda: {"container_id": "abcdef123456", "port": 8000},
+    )
+    monkeypatch.setattr(sync_cmd.ui, "status", lambda _message: contextlib.nullcontext())
+    monkeypatch.setattr(
+        sync_cmd, "SYNC_STATE_PATH", str(tmp_path / "sync-manifest.json")
+    )
+
+    remote_files = {}
+
+    def fake_docker(argv, **_kwargs):
+        if argv[:2] == ["docker", "cp"]:
+            local_path, remote = argv[2:]
+            if remote.endswith(":" + sync_cmd.GC_WORKSPACE_PATH):
+                shutil.copytree(source, destination, dirs_exist_ok=True)
+            else:
+                remote_path = remote.split(":", 1)[1]
+                staged_path = tmp_path / ("remote-" + remote_path.rsplit("/", 1)[-1])
+                shutil.copyfile(local_path, staged_path)
+                remote_files[remote_path] = staged_path
+            return subprocess.CompletedProcess(argv, 0, "", "")
+
+        assert argv[:3] == ["docker", "exec", "abcdef123456"]
+        remote_manifest = remote_files[argv[-1]]
+        return subprocess.run(
+            [sys.executable, "-c", argv[5], str(destination), str(remote_manifest)],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+    monkeypatch.setattr(sync_cmd, "run_docker", fake_docker)
+
+    assert sync_cmd.run_sync()
+    (destination / "managed" / "container-generated.txt").write_text("keep me")
+    (source / "b.txt").unlink()
+    shutil.rmtree(source / "managed")
+    assert sync_cmd.run_sync()
+
+    assert (destination / "a.txt").read_text() == "A"
+    assert not (destination / "b.txt").exists()
+    assert not (destination / "managed" / "source.txt").exists()
+    assert (
+        destination / "managed" / "container-generated.txt"
+    ).read_text() == "keep me"
diff --git a/tests/test_cli.py b/tests/test_cli.py
index f07a39db..16414a5d 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -42,6 +42,7 @@ def test_deploy_skips_ec2_preflight_for_local_config(
         with (
             patch("canyonos_core.cli.os.path.isfile", return_value=True),
             patch("canyonos_core.cli._load_config", return_value=config),
+            patch("canyonos_core.cli.resolve_env_file", return_value=None),
             patch.dict(
                 sys.modules, {"canyonos_core.controller.global_controller": controller_module}
             ),
@@ -75,6 +76,7 @@ def test_deploy_runs_ec2_preflight_for_ec2_config(
         with (
             patch("canyonos_core.cli.os.path.isfile", return_value=True),
             patch("canyonos_core.cli._load_config", return_value=config),
+            patch("canyonos_core.cli.resolve_env_file", return_value=None),
             patch.dict(
                 sys.modules, {"canyonos_core.controller.global_controller": controller_module}
             ),
@@ -101,6 +103,8 @@ def test_deploy_uses_car_when_present(
             "canyonos_core.cli.os.path.isfile", return_value=True
         ), patch(
             "canyonos_core.cli._load_config", return_value={"agents": []}
+        ), patch(
+            "canyonos_core.cli.resolve_env_file", return_value=None
         ), patch.dict(
             sys.modules, {"canyonos_core.controller.global_controller": controller_module}
         ):
@@ -343,7 +347,7 @@ def test_build_fails_when_stub_cannot_be_generated(self):
                 )
             )
 
-            with self.assertRaises(SystemExit):
+            with self.assertRaisesRegex(RuntimeError, "missing `entrypoint`"):
                 self._run_build(project_dir, [], buildx_available=True)
 
     def _write_requirements_config(self, project_dir):
diff --git a/tests/test_config_hardening.py b/tests/test_config_hardening.py
new file mode 100644
index 00000000..ea665c9d
--- /dev/null
+++ b/tests/test_config_hardening.py
@@ -0,0 +1,89 @@
+from pathlib import Path
+
+import pytest
+import yaml
+
+from canyonos_core.cli import _load_config
+
+
+def _write_config(tmp_path, agents):
+    path = Path(tmp_path, "global_controller.yaml")
+    path.write_text(yaml.safe_dump({"agents": agents}))
+    return path
+
+
+@pytest.mark.parametrize("replicas", [0, -1, 1.5, "2", [], True])
+def test_replicas_must_be_a_positive_integer(tmp_path, replicas):
+    path = _write_config(
+        tmp_path,
+        [{"name": "Agent", "entrypoint": "agent.py", "replicas": replicas}],
+    )
+
+    with pytest.raises(
+        RuntimeError, match="positive integer.*replicas|replicas.*positive integer"
+    ):
+        _load_config(path)
+
+
+def test_unknown_provider_is_rejected(tmp_path):
+    path = _write_config(
+        tmp_path,
+        [{"name": "Agent", "entrypoint": "agent.py", "provider": "locla"}],
+    )
+
+    with pytest.raises(RuntimeError, match="unsupported provider"):
+        _load_config(path)
+
+
+def test_provider_case_is_normalized(tmp_path):
+    path = _write_config(
+        tmp_path,
+        [{"name": "Agent", "entrypoint": "agent.py", "provider": "LOCAL"}],
+    )
+
+    assert _load_config(path)["agents"][0]["provider"] == "local"
+
+
+def test_case_colliding_names_are_rejected(tmp_path):
+    path = _write_config(
+        tmp_path,
+        [
+            {"name": "Agent", "entrypoint": "agent.py"},
+            {"name": "agent", "entrypoint": "other.py"},
+        ],
+    )
+
+    with pytest.raises(RuntimeError, match="Duplicate agent names"):
+        _load_config(path)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [("api_port", 0), ("redis_port", 65536), ("host_port", "8000")],
+)
+def test_ports_are_validated(tmp_path, field, value):
+    path = _write_config(
+        tmp_path,
+        [{"name": "Workflow", "workflow_file": "workflow.py", field: value}],
+    )
+
+    with pytest.raises(RuntimeError, match=field):
+        _load_config(path)
+
+
+def test_multiple_local_workflow_replicas_fail_with_an_actionable_error(tmp_path):
+    path = _write_config(
+        tmp_path,
+        [
+            {
+                "name": "Workflow",
+                "type": "workflow",
+                "workflow_file": "workflow.py",
+                "provider": "local",
+                "replicas": 2,
+            }
+        ],
+    )
+
+    with pytest.raises(RuntimeError, match="same `api_port`"):
+        _load_config(path)
diff --git a/tests/test_dashboard_stack.py b/tests/test_dashboard_stack.py
index 77d2600f..19467ffa 100644
--- a/tests/test_dashboard_stack.py
+++ b/tests/test_dashboard_stack.py
@@ -283,8 +283,8 @@ def test_stop_and_teardown_are_a_noop_without_an_env_file(monkeypatch, project):
     calls = []
     install_docker(monkeypatch, calls)
 
-    assert dashboard_stack.stop_dashboard() is False
-    assert dashboard_stack.teardown_dashboard() is False
+    assert dashboard_stack.stop_dashboard() is True
+    assert dashboard_stack.teardown_dashboard() is True
     assert calls == []
 
 
diff --git a/tests/test_deploy_hardening_fixes.py b/tests/test_deploy_hardening_fixes.py
new file mode 100644
index 00000000..dccd6b5b
--- /dev/null
+++ b/tests/test_deploy_hardening_fixes.py
@@ -0,0 +1,149 @@
+"""Regressions for four independent defects found reviewing the CAN-316 diff."""
+
+import os
+import socket
+import subprocess
+import sys
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from canyonos_core.controller.cloud_provider_logic.Local import _runtime as local_runtime
+from canyonos_core.controller.global_controller import GlobalController
+from canyonos_core.controller.instance_manager import InstanceManager
+from canyonos_core.controller.local_controller import LocalController
+
+
+class AgentStartupOrderTests(unittest.TestCase):
+    def test_the_agent_is_constructed_after_the_llm_proxy_is_listening(self):
+        """An agent constructor may build an LLM client against the in-container
+        proxy, so the proxy has to exist first."""
+        order = []
+
+        with patch.dict(
+            os.environ, {"CANYONOS_AGENT_NAME": "Analyst", "CANYONOS_AGENT_PORT": "50051"}
+        ), patch(
+            "canyonos_core.controller.local_controller.start_server",
+            return_value=(MagicMock(), MagicMock()),
+        ), patch(
+            "canyonos_core.controller.local_controller.RedisClient", return_value=MagicMock()
+        ), patch.object(
+            LocalController,
+            "_start_llm_proxy",
+            lambda _self, *_a: order.append("proxy") or MagicMock(),
+        ), patch.object(
+            LocalController,
+            "_load_agent",
+            lambda _self: order.append("agent") or MagicMock(),
+        ):
+            LocalController()
+
+        self.assertEqual(order, ["proxy", "agent"])
+
+
+class WorkflowPortPreflightTests(unittest.TestCase):
+    def test_a_local_port_is_probed_on_the_host_not_our_own_loopback(self):
+        """The controller runs in a container on a bridge network, so its own
+        localhost is not where workflow ports are published."""
+        probed = []
+
+        def fake_connection(address, timeout=None):
+            probed.append(address)
+            raise OSError("refused")
+
+        with patch.object(socket, "create_connection", fake_connection):
+            self.assertFalse(local_runtime._port_check("localhost", 8080))
+            self.assertFalse(local_runtime._port_check("127.0.0.1", 8080))
+
+        self.assertEqual(probed, [(local_runtime.HOST_GATEWAY, 8080)] * 2)
+
+    def test_a_remote_host_is_still_probed_directly(self):
+        probed = []
+
+        def fake_connection(address, timeout=None):
+            probed.append(address)
+            raise OSError("refused")
+
+        with patch.object(socket, "create_connection", fake_connection):
+            self.assertFalse(local_runtime._port_check("10.0.0.5", 8080))
+
+        self.assertEqual(probed, [("10.0.0.5", 8080)])
+
+
+class RedisLaunchRetryTests(unittest.TestCase):
+    def test_a_failed_launch_clears_the_container_name_before_retrying(self):
+        """docker leaves a Created container holding the name when the port bind
+        fails, which would make every retry fail for the wrong reason."""
+        config = {
+            "project_id": "test-project",
+            "agents": [{"name": "First", "host": "localhost", "replicas": 1}],
+        }
+        commands = []
+
+        def fake_run_cmd(_controller, cmd, host, user=None):
+            commands.append(cmd[:])
+            if cmd[:2] == ["docker", "run"]:
+                return subprocess.CompletedProcess(cmd, 125, "", "port is already allocated")
+            return subprocess.CompletedProcess(cmd, 0, "", "")
+
+        with patch.object(GlobalController, "_load_config", return_value=config), patch(
+            "canyonos_core.controller.global_controller.resolve_env_file", return_value=None
+        ), patch(
+            "canyonos_core.controller.global_controller.RedisClient", return_value=MagicMock()
+        ), patch(
+            "canyonos_core.controller.global_controller._wait_for_redis"
+        ), patch(
+            "canyonos_core.controller.global_controller.assign_project_id"
+        ), patch.object(
+            GlobalController, "_database_url", return_value=None
+        ), patch.object(
+            GlobalController, "_redis_container_healthy", return_value=False
+        ), patch.object(
+            GlobalController, "_run_cmd", fake_run_cmd
+        ):
+            with self.assertRaises(RuntimeError) as raised:
+                GlobalController("unused.yaml")
+
+        runs = [cmd for cmd in commands if cmd[:2] == ["docker", "run"]]
+        removes = [cmd for cmd in commands if cmd[:3] == ["docker", "rm", "-f"]]
+        self.assertEqual(len(runs), 3)
+        self.assertGreaterEqual(len(removes), 3)
+        # The real cause survives instead of "name is already in use".
+        self.assertIn("port is already allocated", str(raised.exception))
+
+
+class RuntimeReuseScanTests(unittest.TestCase):
+    def _manager(self, run_cmd):
+        controller = SimpleNamespace(
+            containers={},
+            config={},
+            redis=MagicMock(),
+            node_redis={},
+            _run_cmd=run_cmd,
+        )
+        return InstanceManager(controller, redis_client=MagicMock())
+
+    def test_an_unreachable_host_is_not_reusable_rather_than_fatal(self):
+        def run_cmd(_cmd, host, user=None):
+            raise RuntimeError(f"Command timed out after 180s on {host}: docker inspect")
+
+        manager = self._manager(run_cmd)
+        instance = {"runtime_id": "canyonos-agent-0", "host": "10.0.0.5", "provider": "EC2"}
+
+        with self.assertLogs("canyonos_core.controller.instance_manager", level="WARNING"):
+            self.assertFalse(manager._runtime_is_running(instance))
+
+    def test_a_running_runtime_is_still_reported_as_reusable(self):
+        def run_cmd(_cmd, _host, user=None):
+            return subprocess.CompletedProcess([], 0, "true\n", "")
+
+        manager = self._manager(run_cmd)
+        instance = {"runtime_id": "canyonos-agent-0", "host": "localhost"}
+
+        self.assertTrue(manager._runtime_is_running(instance))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_deploy_progress.py b/tests/test_deploy_progress.py
index 99451731..20752bac 100644
--- a/tests/test_deploy_progress.py
+++ b/tests/test_deploy_progress.py
@@ -119,19 +119,46 @@ def test_a_re_read_ready_line_does_not_double_count():
     )
 
 
-def test_coming_up_short_of_the_announced_replicas_is_not_reported_as_success():
+def test_coming_up_short_of_the_announced_replicas_fails_the_deploy(monkeypatch):
     """`_wait_for_healthy` gives up after its timeout and the controller starts
     anyway, so the up-marker can arrive with agents still unhealthy.
     """
-    tracker, _, _, _ = drive(
-        [
+    monkeypatch.setattr(deploy_cmd, "_deploy_summary", lambda *a: pytest.fail("no summary"))
+    lines = deploy_cmd._queued_lines(
+        iter([
             "INFO:canyonos_core.controller.global_controller:Waiting for 3 replica(s) to become healthy (timeout=300s)...\n",
             "INFO:canyonos_core.controller.global_controller:Controller A (127.0.0.1:1) is ready.\n",
-        ]
+            "Agent B never reached readiness.\n",
+            "INFO:canyonos_core.controller.global_controller:Global controller started, polling every 5s...\n",
+        ])
     )
-    message, all_ready = tracker.agents_ready_message()
-    assert not all_ready
-    assert message == "Workflow up, but only 1/3 agents reported healthy"
+
+    with pytest.raises(RuntimeError, match="only 1/3 agents reported healthy"):
+        deploy_cmd._tail_quiet(
+            lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+            on_ready=lambda _ready: None,
+        )
+
+
+def test_incomplete_readiness_replays_the_buffered_diagnostics(monkeypatch, capsys):
+    monkeypatch.setattr(deploy_cmd, "_deploy_summary", lambda *a: pytest.fail("no summary"))
+    lines = deploy_cmd._queued_lines(
+        iter(
+            [
+                "Waiting for 2 replica(s) to become healthy (timeout=300s)...\n",
+                "Agent B import failed.\n",
+                "Global controller started, polling every 5s...\n",
+            ]
+        )
+    )
+
+    with pytest.raises(RuntimeError, match="only 0/2 agents reported healthy"):
+        deploy_cmd._tail_quiet(
+            lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+            on_ready=lambda _ready: None,
+        )
+
+    assert "Agent B import failed." in capsys.readouterr().out
 
 
 def test_a_run_that_never_announced_replicas_still_reports_ready():
@@ -212,7 +239,10 @@ def test_the_clis_own_status_requests_are_not_shown_or_buffered(monkeypatch):
             ]
         )
     )
-    summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False)
+    summary = deploy_cmd._tail_quiet(
+        lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+        on_ready=lambda _ready: None,
+    )
 
     assert summary == ("url", [])
     assert shown == ["Build complete", "Workflow ready"]
@@ -230,7 +260,61 @@ def test_a_build_that_dies_silently_does_not_hang(monkeypatch, capsys):
     # The queue never yields None: the stream stays open, as it does in reality.
     lines.put = lambda *a, **k: None
 
-    summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False)
-
-    assert summary is None
+    with pytest.raises(RuntimeError, match="stopped or timed out"):
+        deploy_cmd._tail_quiet(
+            lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+            on_ready=lambda _ready: None,
+        )
     assert "Building 2 Docker image(s)" in capsys.readouterr().out
+
+
+def test_a_live_child_that_never_becomes_ready_times_out(monkeypatch):
+    monkeypatch.setattr(deploy_cmd, "_STATUS_POLL_SECONDS", 0.001)
+    monkeypatch.setattr(deploy_cmd, "_STARTUP_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr(deploy_cmd, "_REVEAL_GRACE_SECONDS", 0)
+    monkeypatch.setattr(deploy_cmd, "deploy_status", lambda _p: {"running": True})
+
+    lines = deploy_cmd._queued_lines(iter(()))
+
+    with pytest.raises(RuntimeError, match="stopped or timed out"):
+        deploy_cmd._tail_quiet(
+            lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+            on_ready=lambda _ready: None,
+        )
+
+
+def test_a_log_reader_error_fails_promptly(monkeypatch):
+    class BrokenStream:
+        def __iter__(self):
+            raise OSError("pipe closed")
+
+    monkeypatch.setattr(deploy_cmd, "_STARTUP_TIMEOUT_SECONDS", 10)
+    lines = deploy_cmd._queued_lines(BrokenStream())
+
+    with pytest.raises(RuntimeError, match="Could not read.*pipe closed"):
+        list(deploy_cmd._drain(lines, {"port": 1}))
+
+
+def test_verbose_stops_tailing_once_the_workflow_is_up(monkeypatch):
+    """`-v` prints every line until the workflow is up and then returns --
+    `canyonos logs` re-attaches on demand. Tailing past the summary is what
+    left the old startup deadline silently governing the whole session."""
+    monkeypatch.setattr(
+        deploy_cmd, "_deploy_summary", lambda *_a: ("url", [], "config.yaml")
+    )
+    after_ready = "this line arrives after the workflow is up\n"
+    lines = deploy_cmd._queued_lines(
+        iter([
+            "INFO:canyonos_core:Build complete.\n",
+            "INFO:canyonos_core.controller.global_controller:Global controller started, polling every 5s...\n",
+            after_ready,
+        ])
+    )
+
+    summary = deploy_cmd._tail_verbose(
+        lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False,
+        on_ready=lambda _ready: None,
+    )
+
+    assert summary == ("url", [], "config.yaml")
+    assert lines.get(timeout=1) == after_ready
diff --git a/tests/test_docker_cmd.py b/tests/test_docker_cmd.py
new file mode 100644
index 00000000..68fd75b0
--- /dev/null
+++ b/tests/test_docker_cmd.py
@@ -0,0 +1,36 @@
+import subprocess
+
+from canyonos import docker_cmd
+
+
+def test_run_docker_reports_timeout(monkeypatch):
+    def timed_out(*args, **kwargs):
+        raise subprocess.TimeoutExpired(kwargs["timeout"], args[0])
+
+    monkeypatch.setattr(docker_cmd.subprocess, "run", timed_out)
+    try:
+        docker_cmd.run_docker(["docker", "info"], timeout=3, action="Checking Docker")
+    except RuntimeError as exc:
+        assert str(exc) == "Checking Docker timed out after 3s."
+    else:
+        raise AssertionError("expected timeout error")
+
+
+def test_cleanup_retries_once_and_returns_final_failure(monkeypatch):
+    calls = []
+
+    def failed(argv, **kwargs):
+        calls.append((argv, kwargs["timeout"]))
+        return subprocess.CompletedProcess(argv, 1, "", "busy")
+
+    monkeypatch.setattr(docker_cmd.subprocess, "run", failed)
+    monkeypatch.setattr(docker_cmd.time, "sleep", lambda _: None)
+
+    error = docker_cmd.cleanup_docker(
+        ["docker", "rm", "x"], action="Removing x", timeout=7
+    )
+
+    assert error == "Removing x failed: busy"
+    assert len(calls) == 2
+    assert all(timeout == 7 for _, timeout in calls)
+
diff --git a/tests/test_gc_client_hardening.py b/tests/test_gc_client_hardening.py
new file mode 100644
index 00000000..28c26678
--- /dev/null
+++ b/tests/test_gc_client_hardening.py
@@ -0,0 +1,90 @@
+import io
+import json
+import urllib.error
+
+import pytest
+
+from canyonos import gc
+
+
+class _Response:
+    def __init__(self, body):
+        self.body = body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self):
+        return self.body
+
+
+def test_gc_requests_are_bounded(monkeypatch):
+    seen = {}
+
+    def urlopen(_request, **kwargs):
+        seen.update(kwargs)
+        return _Response(b"{}")
+
+    monkeypatch.setattr(gc.urllib.request, "urlopen", urlopen)
+
+    gc.post_clean(8000)
+
+    assert seen == {"timeout": gc.REQUEST_TIMEOUT_SECONDS}
+
+
+@pytest.mark.parametrize("body", [b"not-json", b"[]", b"null"])
+def test_malformed_success_response_becomes_gc_error(monkeypatch, body):
+    monkeypatch.setattr(gc.urllib.request, "urlopen", lambda *_a, **_k: _Response(body))
+
+    with pytest.raises(gc.GCError, match="invalid"):
+        gc.post_clean(8000)
+
+
+def test_non_mapping_http_error_keeps_the_original_body():
+    error = urllib.error.HTTPError(
+        "http://localhost",
+        500,
+        "error",
+        {},
+        io.BytesIO(json.dumps(["broken"]).encode()),
+    )
+
+    assert gc._error_detail(error) == '["broken"]'
+
+
+def test_endpoint_resolution_errors_are_reported_not_hidden(monkeypatch):
+    """The addresses are display-only, so a failure warns and falls back rather
+    than failing a deploy whose workflow is already live."""
+    warnings = []
+    monkeypatch.setattr(gc.ui, "warn", warnings.append)
+    monkeypatch.setattr(
+        gc, "_request", lambda *_a, **_k: {"workflows": [], "error": "redis down"}
+    )
+
+    assert gc.workflow_endpoints(8000) == []
+    assert any("redis down" in str(warning) for warning in warnings)
+
+
+def test_a_missing_endpoints_route_falls_back_without_warning(monkeypatch):
+    warnings = []
+    monkeypatch.setattr(gc.ui, "warn", warnings.append)
+    monkeypatch.setattr(
+        gc,
+        "_request",
+        lambda *_a, **_k: (_ for _ in ()).throw(gc.GCError("nope", code=404)),
+    )
+
+    assert gc.workflow_endpoints(8000) == []
+    assert warnings == []
+
+
+@pytest.mark.parametrize("body", [b"not-json", b"[]", b"null"])
+def test_malformed_status_response_is_not_reported_as_stopped(monkeypatch, body):
+    """None means "couldn't tell", which the caller counts as a miss -- a
+    verdict of "running" would be the dangerous answer here, not an exception."""
+    monkeypatch.setattr(gc.urllib.request, "urlopen", lambda *_a, **_k: _Response(body))
+
+    assert gc.deploy_status(8000) is None
diff --git a/tests/test_gc_container_name.py b/tests/test_gc_container_name.py
index cc2fefdb..1f307f0c 100644
--- a/tests/test_gc_container_name.py
+++ b/tests/test_gc_container_name.py
@@ -100,9 +100,13 @@ def test_a_running_container_holding_the_name_is_refused_not_removed(docker):
     assert docker.run_calls == []
 
 
-def test_a_port_collision_retry_still_gets_the_name(docker):
+@pytest.mark.parametrize("message", [
+    "Bind for 127.0.0.1:8000 failed: port is already allocated",
+    "listen tcp 127.0.0.1:8000: bind: address already in use",
+])
+def test_a_port_collision_retry_still_gets_the_name(docker, message):
     docker.runs = [
-        completed(["docker", "run"], returncode=125, stderr="Bind for 127.0.0.1:8000 failed: port is already allocated"),
+        completed(["docker", "run"], returncode=125, stderr=message),
         completed(["docker", "run"], stdout=f"{CONTAINER_ID}\n"),
     ]
 
diff --git a/tests/test_gc_server_hardening.py b/tests/test_gc_server_hardening.py
new file mode 100644
index 00000000..b3e46228
--- /dev/null
+++ b/tests/test_gc_server_hardening.py
@@ -0,0 +1,48 @@
+import subprocess
+import time
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from canyonos_core import server
+
+
+def test_clean_has_a_bounded_wait(monkeypatch):
+    process = MagicMock()
+    process.poll.return_value = None
+    process.wait.side_effect = subprocess.TimeoutExpired("controller", 45)
+    monkeypatch.setattr(server, "_gc_process", process)
+
+    response = server.app.test_client().post("/clean")
+
+    assert response.status_code == 504
+    assert "did not stop" in response.get_json()["error"]
+    process.wait.assert_called_once_with(timeout=server.CLEAN_TIMEOUT_SECONDS)
+    assert server._gc_process is process
+
+
+def test_concurrent_deploy_requests_start_only_one_process(monkeypatch, tmp_path):
+    config = tmp_path / "global_controller.yaml"
+    config.write_text("agents: []\n")
+    calls = []
+
+    def popen(*args, **kwargs):
+        calls.append((args, kwargs))
+        time.sleep(0.05)
+        return SimpleNamespace(pid=123, poll=lambda: None)
+
+    monkeypatch.setattr(server, "WORKSPACE_DIR", str(tmp_path))
+    monkeypatch.setattr(server, "_gc_process", None)
+    monkeypatch.setattr(server.subprocess, "Popen", popen)
+
+    def request_deploy():
+        with server.app.test_client() as client:
+            return client.post(
+                "/deploy", json={"config_path": "global_controller.yaml"}
+            ).status_code
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        statuses = sorted(executor.map(lambda _index: request_deploy(), range(2)))
+
+    assert statuses == [200, 409]
+    assert len(calls) == 1
diff --git a/tests/test_global_controller_cleanup.py b/tests/test_global_controller_cleanup.py
index bce8fb35..407d8278 100644
--- a/tests/test_global_controller_cleanup.py
+++ b/tests/test_global_controller_cleanup.py
@@ -312,5 +312,96 @@ def test_running_replica_is_left_alone(self):
         self.assertEqual(controller.removed, [])
 
 
+class ShutdownCleanupTests(unittest.TestCase):
+    def test_agent_removal_failure_does_not_skip_other_agents_or_redis(self):
+        controller = GlobalController.__new__(GlobalController)
+        controller.running = True
+        controller.containers = {"Workflow": ["first", "second"]}
+        controller.controllers = []
+        controller.redis_containers = {
+            "localhost": "canyonos-redis-localhost",
+            "10.0.0.5": "canyonos-redis-10-0-0-5",
+        }
+        controller.node_redis = {host: object() for host in controller.redis_containers}
+
+        class FakeInstanceManager:
+            def __init__(self):
+                self.removals = []
+
+            def list_instances(self):
+                return [{"id": "first"}, {"id": "second"}]
+
+            def _instance_id_from_record(self, instance):
+                return instance["id"]
+
+            def remove_instance(self, instance_id):
+                self.removals.append(instance_id)
+                if instance_id == "first":
+                    raise RuntimeError("remove failed")
+
+        class FakeSupervisor:
+            def __init__(self):
+                self.terminated = False
+
+            def terminate_all(self):
+                self.terminated = True
+
+        controller.instance_manager = FakeInstanceManager()
+        controller.process_supervisor = FakeSupervisor()
+        redis_actions = []
+
+        def run_cmd(cmd, host, user=None):
+            redis_actions.append((host, cmd[1]))
+            return subprocess.CompletedProcess(cmd, 0, "", "")
+
+        controller._run_cmd = run_cmd
+
+        failures = controller.cleanup()
+
+        self.assertTrue(any("agent first: remove failed" in f for f in failures))
+        self.assertEqual(controller.instance_manager.removals, ["first", "second"])
+        self.assertEqual(
+            redis_actions,
+            [
+                ("localhost", "stop"),
+                ("localhost", "rm"),
+                ("10.0.0.5", "stop"),
+                ("10.0.0.5", "rm"),
+            ],
+        )
+        self.assertTrue(controller.process_supervisor.terminated)
+
+    def test_failed_redis_removal_stays_tracked_for_retry(self):
+        controller = GlobalController.__new__(GlobalController)
+        controller.controllers = []
+        controller.redis_containers = {
+            "localhost": "canyonos-redis-localhost",
+            "10.0.0.5": "canyonos-redis-10-0-0-5",
+        }
+        controller.node_redis = {host: object() for host in controller.redis_containers}
+        fail_local_remove = True
+
+        def run_cmd(cmd, host, user=None):
+            if fail_local_remove and host == "localhost" and cmd[1] == "rm":
+                return subprocess.CompletedProcess(cmd, 1, "", "remove failed")
+            return subprocess.CompletedProcess(cmd, 0, "", "")
+
+        controller._run_cmd = run_cmd
+
+        failures = controller._stop_redis_containers()
+
+        self.assertEqual(len(failures), 1)
+        self.assertEqual(
+            controller.redis_containers,
+            {"localhost": "canyonos-redis-localhost"},
+        )
+        self.assertEqual(set(controller.node_redis), {"localhost"})
+
+        fail_local_remove = False
+        self.assertEqual(controller._stop_redis_containers(), [])
+        self.assertEqual(controller.redis_containers, {})
+        self.assertEqual(controller.node_redis, {})
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_global_controller_readiness.py b/tests/test_global_controller_readiness.py
new file mode 100644
index 00000000..20056cda
--- /dev/null
+++ b/tests/test_global_controller_readiness.py
@@ -0,0 +1,32 @@
+from types import SimpleNamespace
+
+import pytest
+
+from canyonos_core.controller.global_controller import GlobalController
+
+
+class _Redis:
+    def get(self, _key):
+        return None
+
+
+def test_unhealthy_replicas_fail_startup_without_entering_the_run_loop(monkeypatch):
+    controller = GlobalController.__new__(GlobalController)
+    controller.instance_manager = SimpleNamespace(
+        list_instances=lambda: [
+            {
+                "agent_name": "BrokenAgent",
+                "host": "127.0.0.1",
+                "host_port": 50051,
+            }
+        ],
+        _routing_endpoint_for=lambda instance: (
+            f"{instance['host']}:{instance['host_port']}"
+        ),
+    )
+    controller.node_redis = {}
+    controller.redis = _Redis()
+    controller._last_status = {}
+
+    with pytest.raises(RuntimeError, match="failed to become healthy.*BrokenAgent"):
+        controller._wait_for_healthy(timeout=0, interval=0)
diff --git a/tests/test_global_controller_redis_reuse.py b/tests/test_global_controller_redis_reuse.py
index 2f3e74a4..20b2236e 100644
--- a/tests/test_global_controller_redis_reuse.py
+++ b/tests/test_global_controller_redis_reuse.py
@@ -48,7 +48,7 @@ def fake_run_cmd(cmd, host, user=None):
 
         return [c for c in run_calls if c[:2] == ["docker", "run"]]
 
-    def test_a_healthy_existing_container_is_reused_not_recreated(self):
+    def test_a_healthy_existing_container_is_reused_without_becoming_owned(self):
         controller = _bare_controller(
             [{"name": "Workflow", "replicas": 1, "redis_port": 6379}]
         )
@@ -58,9 +58,15 @@ def test_a_healthy_existing_container_is_reused_not_recreated(self):
         self.assertEqual(
             docker_run_calls, [], "a healthy existing Redis container must not be recreated"
         )
-        self.assertIn("localhost", controller.redis_containers)
+        self.assertNotIn("localhost", controller.redis_containers)
         self.assertIn("localhost", controller.node_redis)
 
+        controller._run_cmd = MagicMock(
+            return_value=SimpleNamespace(returncode=0, stdout="", stderr="")
+        )
+        self.assertEqual(controller._stop_redis_containers(), [])
+        controller._run_cmd.assert_not_called()
+
     def test_an_unhealthy_or_missing_container_still_gets_created(self):
         controller = _bare_controller(
             [{"name": "Workflow", "replicas": 1, "redis_port": 6379}]
diff --git a/tests/test_global_controller_redis_rollback.py b/tests/test_global_controller_redis_rollback.py
new file mode 100644
index 00000000..84cc8edd
--- /dev/null
+++ b/tests/test_global_controller_redis_rollback.py
@@ -0,0 +1,113 @@
+import os
+import subprocess
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from canyonos_core.controller.global_controller import GlobalController
+
+
+class RedisContainerRollbackTests(unittest.TestCase):
+    def test_constructor_failure_removes_an_already_launched_redis_container(self):
+        config = {
+            "project_id": "test-project",
+            "agents": [
+                {
+                    "name": "First",
+                    "host": "10.0.0.1",
+                    "user": "runner",
+                    "replicas": 1,
+                }
+            ],
+        }
+        calls = []
+        failure = RuntimeError("spec write failed")
+
+        def fake_run_cmd(_controller, cmd, host, user=None):
+            calls.append((host, cmd[:]))
+            if cmd[:2] == ["docker", "inspect"]:
+                return subprocess.CompletedProcess(cmd, 1, "", "not found")
+            return subprocess.CompletedProcess(cmd, 0, "", "")
+
+        with patch.object(GlobalController, "_load_config", return_value=config), patch(
+            "canyonos_core.controller.global_controller.resolve_env_file",
+            return_value=None,
+        ), patch(
+            "canyonos_core.controller.global_controller.RedisClient",
+            return_value=MagicMock(),
+        ), patch(
+            "canyonos_core.controller.global_controller._wait_for_redis"
+        ), patch(
+            "canyonos_core.controller.global_controller.assign_project_id"
+        ), patch.object(
+            GlobalController, "_database_url", return_value=None
+        ), patch.object(
+            GlobalController, "_cleanup_stale_containers"
+        ), patch.object(
+            GlobalController, "_run_cmd", new=fake_run_cmd
+        ), patch(
+            "canyonos_core.controller.global_controller.write_agent_specs",
+            side_effect=failure,
+        ):
+            with self.assertRaisesRegex(RuntimeError, "spec write failed") as raised:
+                GlobalController("config.yaml")
+
+        self.assertIs(raised.exception, failure)
+        self.assertIn(
+            ("10.0.0.1", ["docker", "stop", "canyonos-redis-10-0-0-1"]),
+            calls,
+        )
+        self.assertIn(
+            ("10.0.0.1", ["docker", "rm", "canyonos-redis-10-0-0-1"]),
+            calls,
+        )
+
+    def test_later_launch_failure_rolls_back_an_earlier_owned_container(self):
+        controller = GlobalController.__new__(GlobalController)
+        controller.controllers = [
+            {"name": "First", "host": "10.0.0.1", "user": "runner", "replicas": 1},
+            {"name": "Second", "host": "10.0.0.2", "user": "runner", "replicas": 1},
+        ]
+        controller.redis_containers = {}
+        controller.node_redis = {}
+        controller.redis = None
+        calls = []
+
+        def fake_run_cmd(cmd, host, user=None):
+            calls.append((host, cmd[:]))
+            if cmd[:2] == ["docker", "inspect"]:
+                return subprocess.CompletedProcess(cmd, 1, "", "not found")
+            if cmd[:2] == ["docker", "run"] and host == "10.0.0.2":
+                return subprocess.CompletedProcess(cmd, 1, "", "temporary launch failure")
+            return subprocess.CompletedProcess(cmd, 0, "", "")
+
+        controller._run_cmd = fake_run_cmd
+
+        with patch(
+            "canyonos_core.controller.global_controller.RedisClient",
+            return_value=MagicMock(),
+        ), patch("canyonos_core.controller.global_controller._wait_for_redis"):
+            with self.assertRaisesRegex(
+                RuntimeError, "Failed to launch Redis on 10.0.0.2 after 3 attempts"
+            ):
+                controller._launch_redis_containers()
+
+        second_host_runs = [
+            cmd for host, cmd in calls if host == "10.0.0.2" and cmd[:2] == ["docker", "run"]
+        ]
+        self.assertEqual(len(second_host_runs), 3)
+        self.assertIn(
+            ("10.0.0.1", ["docker", "stop", "canyonos-redis-10-0-0-1"]),
+            calls,
+        )
+        self.assertIn(
+            ("10.0.0.1", ["docker", "rm", "canyonos-redis-10-0-0-1"]),
+            calls,
+        )
+        self.assertEqual(controller.redis_containers, {})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_instance_manager_runtime.py b/tests/test_instance_manager_runtime.py
index a64885a8..38906639 100644
--- a/tests/test_instance_manager_runtime.py
+++ b/tests/test_instance_manager_runtime.py
@@ -60,15 +60,27 @@ def scan_keys(self, pattern):
 
 def _fake_controller():
     redis = _FakeRedis()
+    running = set()
+
+    def fake_run_cmd(cmd, host, user=None):
+        if cmd[:2] == ["docker", "inspect"]:
+            return SimpleNamespace(
+                returncode=0 if cmd[-1] in running else 1,
+                stdout="true\n" if cmd[-1] in running else "",
+            )
+        if cmd[:2] == ["docker", "run"]:
+            running.add(cmd[cmd.index("--name") + 1])
+        elif cmd[:3] == ["docker", "rm", "-f"]:
+            running.discard(cmd[-1])
+        return SimpleNamespace(returncode=0, stdout="")
+
     return SimpleNamespace(
         redis=redis,
         containers={},
         node_redis={},
         redis_containers={},
         config={"poll_interval": 5},
-        # stdout="" (not running) so the orphan-check `docker inspect` probe that now
-        # precedes `docker run` reads a real string instead of erroring on a missing attribute.
-        _run_cmd=MagicMock(return_value=SimpleNamespace(returncode=0, stdout="")),
+        _run_cmd=MagicMock(side_effect=fake_run_cmd),
     )
 
 
@@ -143,6 +155,7 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self):
                 "runtime_id": "canyonos-alpha-0",
             },
         )
+
         self.assertEqual(beta["host"], "localhost")
         self.assertEqual(beta["host_port"], "8001")
         self.assertEqual(
@@ -181,6 +194,17 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self):
             ),
         )
 
+    def test_local_provider_case_is_normalized_before_port_reservation(self):
+        controller = _fake_controller()
+        manager = InstanceManager(controller, controller.redis)
+
+        instance = manager.ensure_instances(
+            [{"name": "Alpha", "provider": "LOCAL"}]
+        )[0]
+
+        self.assertEqual(instance["provider"], "local")
+        self.assertEqual(instance["host_port"], "8000")
+
     def test_bootstrap_instance_passes_poll_interval_env_var(self):
         controller = _fake_controller()
         controller.config = {"poll_interval": 7}
@@ -219,7 +243,7 @@ def test_local_workflow_and_resource_flags_stay_the_same(self):
         controller = _fake_controller()
         manager = InstanceManager(controller, controller.redis)
 
-        with patch.object(local_runtime, "_port_bound", return_value=False):
+        with patch.object(local_runtime, "_port_check", return_value=False):
             manager.ensure_instances(
                 [
                     {
@@ -278,7 +302,7 @@ def test_workflow_bootstrap_fails_fast_on_an_occupied_api_port(self):
         controller = _fake_controller()
         manager = InstanceManager(controller, controller.redis)
 
-        with patch.object(local_runtime, "_port_bound", return_value=True):
+        with patch.object(local_runtime, "_port_check", return_value=True):
             with self.assertRaises(RuntimeError) as ctx:
                 manager.ensure_instances(
                     [{"name": "Workflow", "provider": "local", "type": "workflow"}]
@@ -294,7 +318,7 @@ def test_plain_agent_bootstrap_ignores_api_port_conflicts(self):
         controller = _fake_controller()
         manager = InstanceManager(controller, controller.redis)
 
-        with patch.object(local_runtime, "_port_bound", return_value=True):
+        with patch.object(local_runtime, "_port_check", return_value=True):
             manager.ensure_instances([{"name": "Alpha", "provider": "local"}])
 
         controller._run_cmd.assert_called()
@@ -319,6 +343,58 @@ def test_agent_id_is_published_under_the_controller_endpoint_key(self):
             alpha["agent_id"],
         )
 
+    def test_instance_record_failure_removes_the_new_local_runtime(self):
+        controller = _fake_controller()
+        manager = InstanceManager(controller, controller.redis)
+        write = controller.redis.hset_multiple
+
+        def fail_instance_record(name, mapping):
+            if "runtime_id" in mapping:
+                raise RuntimeError("redis unavailable")
+            write(name, mapping)
+
+        controller.redis.hset_multiple = fail_instance_record
+
+        with self.assertRaisesRegex(RuntimeError, "redis unavailable"):
+            manager.ensure_instances([{"name": "Alpha", "provider": "local"}])
+
+        commands = [call.args[0] for call in controller._run_cmd.call_args_list]
+        self.assertIn(["docker", "rm", "-f", "canyonos-alpha-0"], commands)
+        self.assertEqual(controller.containers["Alpha"], [])
+
+    def test_missing_runtime_behind_stale_record_is_reprovisioned(self):
+        controller = _fake_controller()
+        manager = InstanceManager(controller, controller.redis)
+        key = "agent_instance:local:Alpha:0"
+        stale = {
+            "agent_id": "stale-agent-id",
+            "agent_name": "Alpha",
+            "provider": "local",
+            "replica_index": "0",
+            "host": "localhost",
+            "host_port": "8000",
+            "container_port": "50051",
+            "endpoint": "localhost:8000",
+            "redis_host": "canyonos-redis-localhost",
+            "redis_port": "6379",
+            "runtime_id": "canyonos-alpha-0",
+        }
+        controller.redis.hset_multiple(key, stale)
+        controller.redis.sadd("agent:Alpha:instances", "local:Alpha:0")
+
+        instance = manager.ensure_instances(
+            [{"name": "Alpha", "provider": "local"}]
+        )[0]
+
+        self.assertNotEqual(instance["agent_id"], "stale-agent-id")
+        commands = [call.args[0] for call in controller._run_cmd.call_args_list]
+        self.assertTrue(
+            any(command[:3] == ["docker", "run", "-d"] for command in commands)
+        )
+        self.assertEqual(
+            controller.redis.smembers("agent:Alpha:instances"), {"local:Alpha:0"}
+        )
+
     def test_local_remove_instance_still_removes_the_same_container(self):
         controller = _fake_controller()
         manager = InstanceManager(controller, controller.redis)
diff --git a/tests/test_local_controller_metrics.py b/tests/test_local_controller_metrics.py
index 24989999..7646951c 100644
--- a/tests/test_local_controller_metrics.py
+++ b/tests/test_local_controller_metrics.py
@@ -81,6 +81,34 @@ def smembers(self, key):
 
 
 class LocalControllerMetricsTests(unittest.TestCase):
+    def test_configured_agent_load_failure_never_reports_healthy(self):
+        redis = _FakeRedis()
+        server = MagicMock()
+        servicer = SimpleNamespace(request_queue=None, on_result=None)
+
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "CANYONOS_AGENT_NAME": "MissingAgent",
+                    "CANYONOS_AGENT_FILE": "/definitely/missing-agent.py",
+                },
+            ),
+            patch(
+                "canyonos_core.controller.local_controller.start_server",
+                return_value=(server, servicer),
+            ),
+            patch(
+                "canyonos_core.controller.local_controller.RedisClient",
+                return_value=redis,
+            ),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "Failed to load configured agent"):
+                LocalController()
+
+        self.assertEqual(redis.get("controller:localhost:50051:status"), "failed")
+        server.stop.assert_called_once_with(0)
+
     def test_collect_metrics_returns_expected_keys(self):
         controller = SimpleNamespace(
             _executor=ThreadPoolExecutor(max_workers=1),
diff --git a/tests/test_local_controller_shutdown.py b/tests/test_local_controller_shutdown.py
new file mode 100644
index 00000000..8c9b2194
--- /dev/null
+++ b/tests/test_local_controller_shutdown.py
@@ -0,0 +1,82 @@
+import os
+import sys
+import threading
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")))
+
+from canyonos_core.controller.local_controller import LocalController
+
+
+class LocalControllerShutdownTests(unittest.TestCase):
+    def test_stop_returns_when_executor_work_does_not_finish(self):
+        release_work = threading.Event()
+        work_started = threading.Event()
+        executor = ThreadPoolExecutor(max_workers=1)
+
+        def hang():
+            work_started.set()
+            release_work.wait()
+
+        executor_future = executor.submit(hang)
+        self.assertTrue(work_started.wait(timeout=1))
+        controller = SimpleNamespace(
+            _metrics_stop_event=MagicMock(),
+            _metrics_thread=MagicMock(),
+            _executor=executor,
+            _executor_futures={executor_future: "future-hung"},
+            _executor_futures_lock=threading.Lock(),
+            redis=MagicMock(),
+            _status_key="controller:localhost:50051:status",
+            server=MagicMock(),
+        )
+        stop_finished = threading.Event()
+        stop_errors = []
+
+        def stop_controller():
+            try:
+                LocalController.stop(controller)
+            except BaseException as error:
+                stop_errors.append(error)
+            finally:
+                stop_finished.set()
+
+        try:
+            with (
+                patch(
+                    "canyonos_core.controller.local_controller.EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS",
+                    0.05,
+                ),
+                self.assertLogs(
+                    "canyonos_core.controller.local_controller", level="WARNING"
+                ) as logs,
+            ):
+                started_at = time.monotonic()
+                stop_thread = threading.Thread(target=stop_controller)
+                stop_thread.start()
+                returned_before_deadline = stop_finished.wait(timeout=0.5)
+                elapsed = time.monotonic() - started_at
+                if not returned_before_deadline:
+                    release_work.set()
+                stop_thread.join(timeout=1)
+
+            self.assertTrue(returned_before_deadline)
+            self.assertLess(elapsed, 0.5)
+            self.assertEqual(stop_errors, [])
+            self.assertIn("future-hung", "\n".join(logs.output))
+            controller.redis.set.assert_called_once_with(
+                "controller:localhost:50051:status", "stopped"
+            )
+            controller.server.stop.assert_called_once_with(0)
+        finally:
+            release_work.set()
+            executor_future.result(timeout=1)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/uv.lock b/uv.lock
index 9b817816..5d1fe2a8 100644
--- a/uv.lock
+++ b/uv.lock
@@ -55,7 +55,7 @@ wheels = [
 
 [[package]]
 name = "canyonos"
-version = "0.1.716"
+version = "0.1.717"
 source = { editable = "cli" }
 dependencies = [
     { name = "pyfiglet" },