diff --git a/README.md b/README.md
index d097c241..77246d26 100644
--- a/README.md
+++ b/README.md
@@ -18,9 +18,9 @@ CanyonOS is a control plane that takes your agentic workflow and deploys it, pro
 
 Orchestration: Install Kubernetes/Docker Compose for distributed deployment and management
 
-Observability: Install Langfuse/Arize Phoenix for LLM Observability
+Observability: Install Langfuse/Arize Phoenix for LLM Observability and then Prometheus/Mimir for Instance Metrics
 
-Execution: Install Ray or Kuberay to manage async task execution
+Execution: Install Ray or Kuberay to manage task execution
 
 </td>
 <td valign="top">
@@ -194,6 +194,8 @@ env_file: .env
 
 If you are deploying agents and tools to multiple hosts, make sure the hosts are reachable from the machine running the deploy command and that SSH key-based access is already configured. A guide to set that up can be found [here](https://www.redhat.com/en/blog/passwordless-ssh).
 
+For `provider: EC2` agents specifically (AMI/IAM/security group requirements, the `ec2:` config block), see [canyonos_core/controller/cloud_provider_logic](canyonos_core/controller/cloud_provider_logic/README.md).
+
 ### 6. Stop or quit
 
 ```bash
diff --git a/canyonos_core/Dockerfile b/canyonos_core/Dockerfile
index cd1c3270..5a6b2501 100644
--- a/canyonos_core/Dockerfile
+++ b/canyonos_core/Dockerfile
@@ -2,7 +2,7 @@ FROM python:3.11-slim
 
 ENV PYTHONUNBUFFERED=1
 
-RUN apt-get update && apt-get install -y docker.io && rm -rf /var/lib/apt/lists/*
+RUN apt-get update && apt-get install -y docker.io zstd && rm -rf /var/lib/apt/lists/*
 
 COPY . /src
 RUN pip install /src
diff --git a/canyonos_core/cli.py b/canyonos_core/cli.py
index dd52a2a9..e630a66f 100644
--- a/canyonos_core/cli.py
+++ b/canyonos_core/cli.py
@@ -24,7 +24,6 @@
 ARTIFACT_DIR_NAME = ".car"
 SOURCE_DIR_NAME = "app"
 EC2_REQUIRED_CONFIG_KEYS = (
-    "ami_id",
     "subnet_id",
     "security_group_ids",
     "region",
diff --git a/canyonos_core/controller/cloud_provider_logic/EC2/README.md b/canyonos_core/controller/cloud_provider_logic/EC2/README.md
index ea9ee2b1..2b3afcda 100644
--- a/canyonos_core/controller/cloud_provider_logic/EC2/README.md
+++ b/canyonos_core/controller/cloud_provider_logic/EC2/README.md
@@ -1,48 +1,143 @@
-What you need to run agents on EC2. 
-
-For local controller
-- AMI with the following installed:
-  - Docker
-  - Public key in authorizedkeys in ~/.ssh
-
-For global controller
-- Instance with all of the following installed:
-  - Both things in local controller
-  - Python 3.10+
-  - CanyonOS folder
-  - pip requirements installed in env
-    - pip install -e . --break-system-packages
-  - Private key labeled as ventis_ec2 inside ~/.ssh
-    - Private key complementing local public key
-
-
-For convenience, you can use the global controller as the
-AMI for local to save some time
-
-AWS Specific Permissions
-- IAM Role to attach to global controller instance allowing
-  - ec2:RunInstance,TerminateInstance,DescribeInstances,CreateTags
-- SSH permissions for entering global controller
-- Security group allowing port 50051, 6379, and 22 TCP communication within security group
-- Also need the port your workflow API's are open to be allowed
-
-
-Steps:
-
-1. SSH into EC2 Global Controller Instance
-
-2. Create your app
-
-3. Change the agents/configs/workflow folders to suit your needs
-
-4. Run canyonos build + canyonos deploy
-
-For cleanup, use canyonos clean to clean stubs/containers
-
-If encountering permission errors with the keys, run this to give key more permissions if blocked
-chmod 700 ~/.ssh
-chmod 600 ~/.ssh/ventis_ec2
-
-
-
-
+# EC2 Specific Set-up
+
+The backend for `provider: EC2` agents. It launches one EC2 instance per
+replica, ships the agent's built Docker image to it over SSH, and starts the
+container there. 
+
+## What the host needs to deploy to EC2
+
+** The host means the machine that you are running canyonos deploy on **
+
+- Everything needed for local deployment. (This is Docker and the canyonos CLI)
+- AWS Credentials. The host needs IAM permissions to execute certain EC2 commands. The `AmazonEC2FullAccess` policy in IAM covers all of the needed permissions. For the exact IAM permissions needed, look below in the `IAM Permissions` section.
+- Config block added to `global_controller.yaml` (shown below). The security group would also need to allow specific ports in.
+
+## AWS-side setup
+
+- **Security group** (everything is inbound, same permissions needed for outbound unless you allow all outbound traffic)
+  - The workflow and dashboard port need to be opened up, they are by default 8080 and 8081, and are referred as such below
+  - Port 50051, 6379, and 22 needs to be accessible by the security group, as well as wherever the host runs.
+  - Port 8080 needs be accessible by whoever queries the workflow (0.0.0.0 for example)
+
+## Config (`ec2:` block in `global_controller.yaml`)
+
+You need to add this block to global_controller.yaml, to give canyonos the necessary credentials to launch agents on EC2.
+
+```yaml
+ec2:
+  region: us-east-1
+  subnet_id: subnet-0123456789abcdef0
+  security_group_ids:
+    - sg-0123456789abcdef0
+```
+
+`canyonos build` handles this automatically, but each EC2 agent's spec also needs its own `instance_type` (e.g. `t3.micro`), example below. 
+
+Example:
+
+```yaml
+  - name: ExampleAgent
+    entrypoint: agents/example_agent.py
+    provider: EC2
+    instance_type: t3.micro
+```
+
+## IAM permissions
+
+The host machine needs these IAM Permissions to be able to launch external EC2 instances. This can either be held by the host itself, or the EC2 machine hosting the deployment.
+
+- ec2:
+  - RunInstances
+  - TerminateInstances
+  - DescribeInstances
+  - CreateTags
+  - ImportKeyPair
+- IAM:
+  - PassRole (only needed if you set `ec2.instance_profile_name`)
+
+You can paste the exact JSON below in the "create manual policy" field in IAM when generating permissions for a role. Drop the `iam:PassRole` statement if you're not setting `ec2.instance_profile_name`.
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "ec2:RunInstances",
+        "ec2:TerminateInstances",
+        "ec2:DescribeInstances",
+        "ec2:CreateTags",
+        "ec2:ImportKeyPair"
+      ],
+      "Resource": "*"
+    }
+  ]
+}
+```
+
+
+## [OPTIONAL] Extra Configurations in the ec2 block in `global_controller.yaml`
+
+The ec2 config block given above has the bare minimum needed to get the instances up and running, to add separate settings, you can add these following fields in the block.
+
+```yaml
+ec2:
+  # Required Commands
+  region: us-east-1
+  subnet_id: subnet-0123456789abcdef0
+  security_group_ids:
+    - sg-0123456789abcdef0
+
+  # Optional Commands
+  instance_profile_name: ec2launch # For the launched instances, if you want a specific IAM role attached to each instance, pass this variable in.
+  ami_id: ami-0123456789abcdef0 # Look at `Creating your own AMI` below
+  ssh_user: ubuntu # Look at `Creating your own AMI` below
+  ssh_private_key_path: ~/.ssh/your-own-key # Look at `Private Key` below
+```
+For the instance_profile_name, you will also need to add a new policy to your IAM role in the host machine, the iam:PassRole policy. JSON below.
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": "iam:PassRole",
+      "Resource": "arn:aws:iam::<account-id>:role/<role-behind-your-instance-profile>"
+    }
+  ]
+}
+```
+
+## [OPTIONAL] Private Key
+
+`ec2.ssh_private_key_path` is optional, defaulting to `~/.ssh/canyonos_ec2`.
+If nothing exists at that default path, canyonos generates a fresh ed25519
+keypair there for you on first use (readable only by its owner: `chmod 600`).
+If you set `ec2.ssh_private_key_path` to your own path instead, that file
+must already exist — canyonos only auto-generates the default, never a
+path you explicitly configured.
+
+The AWS-side key pair is always automatic either way: canyonos imports
+whichever key you end up with (generated or your own) into AWS on first use,
+named `canyonos-ec2-<project_id>-<pubkey hash>`, so the AMI never needs the
+key pre-authorized.
+
+## [OPTIONAL] Creating your own AMI
+
+We have provided our own base AMI_ID: ami-0101d5f2a2a9cd55c, but if you want to create your own, the ami you create just needs to have docker and zstd installed. The commands to install it are below.
+If using a different AMI base than Ubuntu, you will need to change the ssh_user manually.
+
+```bash
+
+#!/bin/bash
+set -eux
+
+apt-get update
+apt-get install -y docker.io zstd
+
+systemctl enable docker
+systemctl start docker
+
+```
\ No newline at end of file
diff --git a/canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py b/canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py
index 64f6e1d0..71f3303d 100644
--- a/canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py
+++ b/canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py
@@ -12,16 +12,19 @@
 they can read config and reuse the controller's Docker/Redis logic.
 """
 
+import hashlib
 import logging
 import os
 import shlex
 import socket
 import stat
 import subprocess
+import threading
 import time
 from typing import Any
 
 import boto3
+from botocore.exceptions import ClientError
 
 from canyonos_core.controller.utils.container_names import container_name
 from canyonos_core.controller.utils.env_file import env_file_args
@@ -32,18 +35,46 @@
 
 CONTAINER_PORT = 50051
 PROVIDER = "ec2"
-DEFAULT_SSH_KEY_PATH = os.path.expanduser("~/.ssh/ventis_ec2")
+PUBLIC_IP_TIMEOUT = 120
+CONTROLLER_HEALTH_TIMEOUT = 180
+DEFAULT_SSH_USER = "ubuntu"
+DEFAULT_SSH_KEY_PATH = os.path.expanduser("~/.ssh/canyonos_ec2")
+
+# This default AMI is a public AMI created by Canyon Code, containing base Ubuntu + Docker + zstd. Can be overriden manually with your own ami_id
+DEFAULT_AMI_ID = "ami-0101d5f2a2a9cd55c"
 _controller: Any = None
+_default_key_lock = threading.Lock()
+
+
+def _generate_default_key(key_path):
+    """Create a fresh ed25519 keypair at the default SSH key path.
+
+    Guarded by a lock since replicas provision concurrently in a thread pool
+    -- without it, two threads could both see the file missing and race to
+    generate it at the same path.
+    """
+    with _default_key_lock:
+        if os.path.isfile(key_path):
+            return
+        os.makedirs(os.path.dirname(key_path), exist_ok=True, mode=0o700)
+        subprocess.run(
+            ["ssh-keygen", "-t", "ed25519", "-N", "", "-f", key_path, "-q"],
+            check=True,
+        )
+        os.chmod(key_path, 0o600)
 
 
 def _ssh_key_path(cfg):
     """Return the configured EC2 SSH identity after validating it locally."""
-    key_path = os.path.expanduser(cfg.get("ssh_private_key_path", DEFAULT_SSH_KEY_PATH))
+    configured_path = cfg.get("ssh_private_key_path")
+    key_path = os.path.expanduser(configured_path or DEFAULT_SSH_KEY_PATH)
     if not os.path.isfile(key_path):
-        raise ValueError(
-            f"EC2 SSH private key does not exist: {key_path}. "
-            "Set ec2.ssh_private_key_path to the controller key."
-        )
+        if configured_path:
+            raise ValueError(
+                f"EC2 SSH private key does not exist: {key_path}. "
+                "Set ec2.ssh_private_key_path to the controller key."
+            )
+        _generate_default_key(key_path)
     if not os.access(key_path, os.R_OK):
         raise ValueError(f"EC2 SSH private key is not readable: {key_path}")
 
@@ -59,18 +90,45 @@ def _ssh_key_path(cfg):
 def _aws_clients():
     """Return validated EC2 config and EC2 client."""
     cfg = _controller.config.get("ec2", {})
+    cfg.setdefault("ami_id", DEFAULT_AMI_ID)
+    cfg.setdefault("ssh_user", DEFAULT_SSH_USER)
     required = [
-        "ami_id",
         "subnet_id",
         "security_group_ids",
         "region",
-        "ssh_user",
     ]
     missing = [field for field in required if not cfg.get(field)]
     if missing:
         raise ValueError(f"Missing EC2 config: {', '.join(sorted(missing))}")
-    _ssh_key_path(cfg)
-    return cfg, boto3.client("ec2", region_name=cfg["region"])
+    key_path = _ssh_key_path(cfg)
+    client = boto3.client("ec2", region_name=cfg["region"])
+    _ensure_key_pair_imported(cfg, key_path, client)
+    return cfg, client
+
+
+def _ensure_key_pair_imported(cfg, key_path, client):
+    """Import the controller's SSH public key as a per-project EC2 key pair, if it isn't already.
+
+    Naming the key pair after the project id and a hash of the public key itself
+    (rather than one fixed name) keeps unrelated projects, and a key that gets
+    regenerated after the original file is lost, from colliding under the same
+    AWS key pair name and silently drifting out of sync with it.
+    """
+    public_key = subprocess.run(
+        ["ssh-keygen", "-y", "-f", key_path],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.encode()
+    project_id = _controller.config.get("project_id") or "default"
+    fingerprint = hashlib.sha256(public_key).hexdigest()[:8]
+    key_name = f"canyonos-ec2-{project_id}-{fingerprint}"
+    try:
+        client.import_key_pair(KeyName=key_name, PublicKeyMaterial=public_key)
+    except ClientError as exc:
+        if exc.response.get("Error", {}).get("Code") != "InvalidKeyPair.Duplicate":
+            raise
+    cfg["key_pair_name"] = key_name
 
 
 def provision_instance(spec, replica_index, next_host_port=None):
@@ -82,9 +140,9 @@ def provision_instance(spec, replica_index, next_host_port=None):
         "InstanceType": spec["instance_type"],
         "SubnetId": cfg["subnet_id"],
         "SecurityGroupIds": cfg["security_group_ids"],
+        "KeyName": cfg["key_pair_name"],
         "MinCount": 1,
         "MaxCount": 1,
-        "IamInstanceProfile": {"Name": "ec2launch"},
         "TagSpecifications": [
             {
                 "ResourceType": "instance",
@@ -104,13 +162,15 @@ def provision_instance(spec, replica_index, next_host_port=None):
             },
         ],
     }
+    if cfg.get("instance_profile_name"):
+        request["IamInstanceProfile"] = {"Name": cfg["instance_profile_name"]}
 
     response = client.run_instances(**request)
     instance_id = response["Instances"][0]["InstanceId"]
     runtime_id = f"{container_name(agent_name, replica_index)}--{instance_id}"
     client.get_waiter("instance_running").wait(InstanceIds=[instance_id])
 
-    deadline = time.time() + cfg.get("public_ip_timeout", 120)
+    deadline = time.time() + PUBLIC_IP_TIMEOUT
     instance = None
     while time.time() < deadline:
         response = client.describe_instances(InstanceIds=[instance_id])
@@ -122,13 +182,13 @@ def provision_instance(spec, replica_index, next_host_port=None):
             if instance:
                 break
         if instance and (
-            instance.get("PrivateIpAddress") or instance.get("PublicIpAddress")
+            instance.get("PublicIpAddress") or instance.get("PrivateIpAddress")
         ):
             break
         time.sleep(2)
 
     host = (
-        instance.get("PrivateIpAddress") or instance.get("PublicIpAddress")
+        instance.get("PublicIpAddress") or instance.get("PrivateIpAddress")
         if instance
         else None
     )
@@ -136,8 +196,10 @@ def provision_instance(spec, replica_index, next_host_port=None):
         raise RuntimeError(
             f"EC2 instance {instance_id} does not have a reachable IP address."
         )
-    # Kept alongside `host` (a private IP inside the VPC): callers outside the
-    # VPC, such as the CLI printing where to send requests, need this one.
+    # Public preferred: the deploying host (the machine running `canyonos
+    # deploy`) is outside the VPC in the common case, so SSH/build traffic
+    # needs the public IP. Kept alongside `host` for callers that want it
+    # explicitly regardless of which one `host` ended up being.
     public_host = instance.get("PublicIpAddress") if instance else None
 
     redis_port = spec.get(
@@ -171,10 +233,7 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
             redis_port=redis_port,
             agent_id=agent_id,
         )
-        _check_controller_health(
-            f"{host}:{CONTAINER_PORT}",
-            timeout=cfg.get("controller_health_timeout", 180),
-        )
+        _check_controller_health(f"{host}:{CONTAINER_PORT}")
         instance = {
             "agent_name": spec["name"],
             "provider": "EC2",
@@ -325,13 +384,10 @@ def _bootstrap_instance(
         )
 
 
-def _check_controller_health(endpoint, timeout=None):
+def _check_controller_health(endpoint, timeout=CONTROLLER_HEALTH_TIMEOUT):
     """Wait until the launched container accepts TCP connections."""
     host, port = endpoint.split(":")
-    deadline = time.time() + (
-        timeout
-        or _controller.config.get("ec2", {}).get("controller_health_timeout", 180)
-    )
+    deadline = time.time() + timeout
     while time.time() < deadline:
         try:
             with socket.create_connection((host, int(port)), timeout=2):
diff --git a/canyonos_core/controller/cloud_provider_logic/Local/README.md b/canyonos_core/controller/cloud_provider_logic/Local/README.md
new file mode 100644
index 00000000..5e6037ee
--- /dev/null
+++ b/canyonos_core/controller/cloud_provider_logic/Local/README.md
@@ -0,0 +1,6 @@
+# Local
+
+The default backend for `provider: local` agents. Runs each agent as a Docker
+container on the same machine where you run `canyonos deploy`.
+
+The deployment steps are automatically covered in the root README.md.
diff --git a/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py b/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
index 5f1c7e43..02bac4f3 100644
--- a/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
+++ b/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py
@@ -8,7 +8,6 @@
 
 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
@@ -29,39 +28,22 @@ def _require_controller():
     return _controller
 
 
-def _is_local_host(host):
-    return host in {"localhost", "127.0.0.1"}
-
-
-def _port_bound(port):
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
-        # docker publishes ports with SO_REUSEADDR, so a lingering TIME_WAIT
-        # socket doesn't stop it. Match that, or the probe reports a port as
-        # taken right after a container that used it went down.
-        probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        try:
-            probe.bind(("127.0.0.1", port))
-        except OSError:
-            return True
-    return False
-
-
 def validate_config():
     return None
 
 
 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))))
+    host_port = int(
+        spec.get("host_port", spec.get("port", next_host_port(DEFAULT_HOST)))
+    )
     agent_name = spec["name"]
 
     return {
         "provider": PROVIDER,
-        "host": host,
+        "host": DEFAULT_HOST,
         "host_port": host_port,
-        "redis_host": f"canyonos-redis-{host.replace('.', '-')}",
+        "redis_host": f"canyonos-redis-{DEFAULT_HOST}",
         "runtime_id": container_name(agent_name, replica_index),
-        "user": spec.get("user"),
     }
 
 
@@ -70,14 +52,14 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
     resources = spec.get("resources", {})
     ctrl_type = spec.get("type", "agent")
     image = f"canyonos-{agent_name.lower()}"
-    host = provisioned["host"]
     host_port = provisioned["host_port"]
-    user = provisioned.get("user")
     redis_host = provisioned["redis_host"]
     runtime_id = provisioned["runtime_id"]
 
     inspect = _require_controller()._run_cmd(
-        ["docker", "inspect", "-f", "{{.State.Running}}", runtime_id], host, user
+        ["docker", "inspect", "-f", "{{.State.Running}}", runtime_id],
+        DEFAULT_HOST,
+        None,
     )
     if inspect.returncode == 0 and inspect.stdout.strip() == "true":
         logger.warning(
@@ -85,15 +67,9 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
             "treating it as orphaned and recreating.",
             runtime_id,
         )
-        _require_controller()._run_cmd(["docker", "rm", "-f", runtime_id], host, user)
-
-    if ctrl_type == "workflow" and _is_local_host(host):
-        api_port = int(spec.get("api_port", 8080))
-        if _port_bound(api_port):
-            raise RuntimeError(
-                f"api_port {api_port} is already in use and can't be reassigned "
-                "automatically -- free it or change `api_port` in the workflow's config."
-            )
+        _require_controller()._run_cmd(
+            ["docker", "rm", "-f", runtime_id], DEFAULT_HOST, None
+        )
 
     for attempt in range(MAX_PORT_ATTEMPTS):
         cmd = [
@@ -129,12 +105,10 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         # value, or empty -- so it ALWAYS wins over --env-file (docker: -e beats
         # --env-file). A user's .env can therefore neither enable the stub nor
         # change it; it is reachable only through `canyonos test`.
-        cmd.extend(
-            [
-                "-e",
-                f"CANYONOS_LLM_STUB_TEXT={os.environ.get('CANYONOS_LLM_STUB_TEXT', '')}",
-            ]
-        )
+        cmd.extend([
+            "-e",
+            f"CANYONOS_LLM_STUB_TEXT={os.environ.get('CANYONOS_LLM_STUB_TEXT', '')}",
+        ])
 
         if ctrl_type == "workflow":
             cmd.extend(["-p", f"{spec.get('api_port', 8080)}:8080"])
@@ -155,11 +129,11 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         # User secrets from `env_file`. Explicit -e flags above still win, so a
         # stray CANYONOS_* line in someone's .env cannot break agent wiring.
         with env_file_args(
-            _require_controller(), host, user, runtime_id, _is_local_host(host)
+            _require_controller(), DEFAULT_HOST, None, runtime_id, True
         ) as env_args:
             cmd.extend(env_args)
             cmd.append(image)
-            result = _require_controller()._run_cmd(cmd, host, user)
+            result = _require_controller()._run_cmd(cmd, DEFAULT_HOST, None)
 
         if result.returncode == 0:
             break
@@ -169,7 +143,7 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
             # retrying with a new port, or the retry hits a name conflict
             # instead of the port conflict we're trying to work around.
             _require_controller()._run_cmd(
-                ["docker", "rm", "-f", runtime_id], host, user
+                ["docker", "rm", "-f", runtime_id], DEFAULT_HOST, None
             )
             host_port += 1
             continue
@@ -187,16 +161,14 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         "agent_name": agent_name,
         "provider": PROVIDER,
         "replica_index": str(replica_index),
-        "host": host,
+        "host": DEFAULT_HOST,
         "host_port": str(host_port),
         "container_port": str(CONTAINER_PORT),
-        "endpoint": f"{host}:{host_port}",
+        "endpoint": f"{DEFAULT_HOST}:{host_port}",
         "redis_host": redis_host,
         "redis_port": str(spec.get("redis_port", 6379)),
         "runtime_id": runtime_id,
     }
-    if user:
-        instance["user"] = user
     if ctrl_type == "workflow":
         instance["api_port"] = str(spec.get("api_port", 8080))
     logger.info("Runtime ready: %s -> %s", runtime_id, instance["endpoint"])
@@ -210,8 +182,8 @@ def terminate_instance(instance):
 
     result = _require_controller()._run_cmd(
         ["docker", "rm", "-f", runtime_id],
-        instance.get("host", DEFAULT_HOST),
-        instance.get("user"),
+        DEFAULT_HOST,
+        None,
     )
     if result.returncode != 0:
         logger.warning("Failed to remove runtime %s", runtime_id)
diff --git a/canyonos_core/controller/cloud_provider_logic/README.md b/canyonos_core/controller/cloud_provider_logic/README.md
new file mode 100644
index 00000000..1da06479
--- /dev/null
+++ b/canyonos_core/controller/cloud_provider_logic/README.md
@@ -0,0 +1,15 @@
+# Cloud Provider Logic
+
+Backend implementations for where CanyonOS runs an agent's container. Every agent
+in `global_controller.yaml` picks one via `provider: local` or `provider: EC2`.
+
+What each provider differs in is where the launched agent lives in, local lives in the host terminal and EC2 would spawn in another EC2 instance.
+
+## Providers
+
+| Provider | Folder | Compute |
+| --- | --- | --- |
+| `local` (default) | `Local/` | Docker container on the same machine, the root README has the whole flow. |
+| `EC2` | `EC2/` | One EC2 instance per replica. |
+
+See [EC2/README.md](EC2/README.md) for EC2-specific setup.
diff --git a/canyonos_core/controller/global_controller.py b/canyonos_core/controller/global_controller.py
index 7e250636..3f388966 100644
--- a/canyonos_core/controller/global_controller.py
+++ b/canyonos_core/controller/global_controller.py
@@ -173,7 +173,8 @@ def _cleanup_stale_containers(self):
         host_containers = {}
 
         for ctrl in self.controllers:
-            user = ctrl.get("user")
+            is_ec2 = ctrl.get("provider", "local").upper() == "EC2"
+            user = ctrl.get("user") if is_ec2 else None
             placements = self._get_replica_placements(ctrl)
 
             for i, (host, port) in enumerate(placements):
@@ -294,14 +295,18 @@ def _write_otel_destinations(self, destinations):
     def _get_replica_placements(ctrl):
         """Normalize replicas into a list of (host, port) placements."""
         replicas = ctrl.get("replicas", 1)
-        default_host = ctrl.get("host", "localhost")
+        is_ec2 = ctrl.get("provider", "local").upper() == "EC2"
+        default_host = ctrl.get("host", "localhost") if is_ec2 else "localhost"
         base_port = ctrl.get("port", 50051)
 
         if isinstance(replicas, int):
             return [(default_host, base_port + i) for i in range(replicas)]
         if isinstance(replicas, list):
             return [
-                (r.get("host", default_host), r.get("port", base_port))
+                (
+                    r.get("host", default_host) if is_ec2 else default_host,
+                    r.get("port", base_port),
+                )
                 for r in replicas
             ]
         return [(default_host, base_port)]
@@ -426,7 +431,8 @@ def _launch_redis_containers(self):
         # Collect unique nodes from all replica placements
         nodes = {}
         for ctrl in self.controllers:
-            user = ctrl.get("user")
+            is_ec2 = ctrl.get("provider", "local").upper() == "EC2"
+            user = ctrl.get("user") if is_ec2 else None
             redis_port = ctrl.get("redis_port", 6379)
             for host, _port in self._get_replica_placements(ctrl):
                 if host not in nodes:
@@ -511,19 +517,10 @@ def _launch_redis_containers(self):
 
     def _stop_redis_containers(self):
         """Stop and remove all launched Redis containers."""
-        nodes = {}
-        for ctrl in self.controllers:
-            if ctrl.get("provider", "local").upper() == "EC2":
-                continue
-            user = ctrl.get("user")
-            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():
-            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)
+                self._run_cmd(["docker", "stop", container_name], host, None)
+                self._run_cmd(["docker", "rm", container_name], 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)
@@ -848,7 +845,7 @@ def _send(instance):
     def _ssh_args(self, host, user=None):
         """Return the `ssh ... target` prefix used to reach a remote host."""
         ssh_key_path = os.path.expanduser(
-            self.config.get("ec2", {}).get("ssh_private_key_path", "~/.ssh/ventis_ec2")
+            self.config.get("ec2", {}).get("ssh_private_key_path", "~/.ssh/canyonos_ec2")
         )
         return [
             "ssh",
diff --git a/canyonos_core/controller/instance_manager.py b/canyonos_core/controller/instance_manager.py
index 52dab241..9665290e 100644
--- a/canyonos_core/controller/instance_manager.py
+++ b/canyonos_core/controller/instance_manager.py
@@ -59,9 +59,12 @@ def ensure_instances(self, agent_specs):
 
                 reserved_port = None
                 if provider == "local":
-                    host = agent_spec.get("host", local_runtime.DEFAULT_HOST)
                     reserved_port = self._next_host_port(
-                        host, key, agent_name, provider, replica_index
+                        local_runtime.DEFAULT_HOST,
+                        key,
+                        agent_name,
+                        provider,
+                        replica_index,
                     )
 
                 jobs.append(
diff --git a/canyonos_core/controller/utils/redis_client.py b/canyonos_core/controller/utils/redis_client.py
index 9b52d8ee..e5eca893 100644
--- a/canyonos_core/controller/utils/redis_client.py
+++ b/canyonos_core/controller/utils/redis_client.py
@@ -1,11 +1,29 @@
+import socket
+
 import redis
 
 
+def _prefer_ipv4(host):
+    """Resolve `host` to an IPv4 address when one exists.
+
+    Some hosts (e.g. Docker Desktop's host.docker.internal) advertise both an
+    IPv4 and an unreachable IPv6 address for the same name; redis-py has no
+    address-family fallback, so it fails outright if it happens to try the
+    broken one first. Falls back to the original host if resolution fails
+    (already an IP, unresolvable, IPv4-only environments, etc).
+    """
+    try:
+        infos = socket.getaddrinfo(host, None, socket.AF_INET)
+    except socket.gaierror:
+        return host
+    return infos[0][4][0] if infos else host
+
+
 class RedisClient(object):
     """Redis utility for connecting to localhost with support for strings, hashes, and sets."""
 
     def __init__(self, host="localhost", port=6379, db=0):
-        self.client = redis.Redis(host=host, port=port, db=db)
+        self.client = redis.Redis(host=_prefer_ipv4(host), port=port, db=db)
 
     # --- String operations ---
 
diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py
index 361a1116..c5346749 100644
--- a/cli/canyonos/deploy.py
+++ b/cli/canyonos/deploy.py
@@ -299,7 +299,7 @@ def _summary_body(dashboard_url, targets, config_path):
         body.append("\n")
         body.append(_curl_example(f"{base}/{route}", example), WHITE)
         body.append("\npoll       ", "dim")
-        body.append(f"{base}/status/<request_id>", WHITE)
+        body.append(f"curl {base}/status/<request_id>", WHITE)
         if host not in ("127.0.0.1", "localhost"):
             body.append(f"\n           needs inbound TCP {port} open on {host}", "dim")
     return body
diff --git a/examples/helloworld/config/global_controller.yaml b/examples/helloworld/config/global_controller.yaml
index 8b24a496..a3643ca9 100644
--- a/examples/helloworld/config/global_controller.yaml
+++ b/examples/helloworld/config/global_controller.yaml
@@ -55,4 +55,4 @@ ec2:
     - sg-0123456789abcdef0
   ssh_user: ubuntu
   # Private key on the global controller used to SSH to EC2 workers.
-  ssh_private_key_path: ~/.ssh/ventis_ec2
+  ssh_private_key_path: ~/.ssh/canyonos_ec2
diff --git a/examples/portfolio/.env.example b/examples/portfolio/.env.example
index b8e6c8fe..8d3a531a 100644
--- a/examples/portfolio/.env.example
+++ b/examples/portfolio/.env.example
@@ -12,6 +12,11 @@ AWS_BEARER_TOKEN_BEDROCK=
 # .claude/skills/porting-to-canyonos/references/llm-proxy.md.
 AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://host.docker.internal:8081/bedrock
 
+# --- AWS EC2 (provider: EC2 agents, config/global_controller.yaml's ec2: block) ---
+EC2_REGION=
+EC2_SUBNET_ID=
+EC2_SECURITY_GROUP_ID=
+
 # --- CanyonOS controller / web ---
 CANYONOS_API_IMAGE=
 CANYONOS_WEB_IMAGE=
diff --git a/examples/portfolio/config/global_controller.yaml b/examples/portfolio/config/global_controller.yaml
index 8e6ae1aa..e2530497 100644
--- a/examples/portfolio/config/global_controller.yaml
+++ b/examples/portfolio/config/global_controller.yaml
@@ -16,7 +16,7 @@ agents:
       cpu: 1
       memory: 256
     entrypoint: agents/intent_agent.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
 
   # Stage 0b: price history fetch. Network/IO-bound, cheap CPU. Called by
@@ -28,7 +28,7 @@ agents:
       cpu: 1
       memory: 256
     entrypoint: agents/price_agent.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
     requirements: [yfinance]
 
@@ -41,7 +41,7 @@ agents:
       cpu: 1
       memory: 256
     entrypoint: agents/metrics_agent.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
 
   # Stage 2: portfolio-level risk aggregation. Single call per request; needs
@@ -53,7 +53,7 @@ agents:
       cpu: 1
       memory: 256
     entrypoint: agents/risk_agent.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
 
   # Stage 3: LLM briefing via Bedrock. On the critical path, one call per
@@ -65,7 +65,7 @@ agents:
       cpu: 1
       memory: 512
     entrypoint: agents/advisor_agent.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
 
   # The workflow, exposed as a REST API.
@@ -76,7 +76,7 @@ agents:
     redis_port: 6379
     replicas: 1
     workflow_file: workflow/portfolio_workflow.py
-    provider: local
+    provider: Local
     instance_type: t3.micro
 
 
@@ -93,6 +93,10 @@ otel:
 # Polling interval in seconds
 poll_interval: 4
 
+# Hands each agent container the project's .env (AWS credentials, Bedrock
+# bearer token, etc.) via `docker run --env-file`.
+env_file: .env
+
 redis:
   host: localhost
   port: 6379
@@ -101,10 +105,7 @@ redis:
 # EC2 defaults for `provider: EC2` replicas.
 ec2:
   region: ${EC2_REGION}
-  ami_id: ${EC2_AMI_ID}
   subnet_id: ${EC2_SUBNET_ID}
   security_group_ids:
     - ${EC2_SECURITY_GROUP_ID}
-  ssh_user: ${EC2_SSH_USER}
-  ssh_private_key_path: ${EC2_SSH_PRIVATE_KEY_PATH}
 
diff --git a/tests/test_global_controller_cleanup.py b/tests/test_global_controller_cleanup.py
index 597fec67..74267f4d 100644
--- a/tests/test_global_controller_cleanup.py
+++ b/tests/test_global_controller_cleanup.py
@@ -268,8 +268,10 @@ def _controller(agents, running=False):
         controller.controllers = agents
         controller.instance_manager = InstanceManager(controller)
         controller.removed = []
+        controller.command_targets = []
 
         def run_cmd(cmd, host, user=None):
+            controller.command_targets.append((host, user))
             if cmd[:2] == ["docker", "inspect"] and running:
                 return subprocess.CompletedProcess(cmd, 0, "true\n", "")
             if cmd[:2] == ["docker", "rm"]:
@@ -317,6 +319,21 @@ def test_running_replica_is_left_alone(self):
 
         self.assertEqual(controller.removed, [])
 
+    def test_local_cleanup_ignores_host_and_user_overrides(self):
+        agents = [
+            {
+                "name": "IntentAgent",
+                "host": "10.0.0.5",
+                "user": "ubuntu",
+                "replicas": [{"host": "10.0.0.6", "port": 50061}],
+            }
+        ]
+        controller = self._controller(agents)
+
+        controller._cleanup_stale_containers()
+
+        self.assertEqual(set(controller.command_targets), {("localhost", None)})
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_global_controller_redis_reuse.py b/tests/test_global_controller_redis_reuse.py
index bf0810d1..f78a52ab 100644
--- a/tests/test_global_controller_redis_reuse.py
+++ b/tests/test_global_controller_redis_reuse.py
@@ -80,7 +80,7 @@ def test_an_unhealthy_or_missing_container_still_gets_created(self):
         )
         self.assertIn("localhost", controller.redis_containers)
 
-    def test_reuse_check_probes_the_exact_expected_container_name(self):
+    def test_local_reuse_check_ignores_remote_host_and_user_overrides(self):
         controller = _bare_controller(
             [
                 {
@@ -88,6 +88,7 @@ def test_reuse_check_probes_the_exact_expected_container_name(self):
                     "replicas": 1,
                     "redis_port": 6379,
                     "host": "10.0.0.5",
+                    "user": "ubuntu",
                 }
             ]
         )
@@ -96,7 +97,7 @@ def test_reuse_check_probes_the_exact_expected_container_name(self):
 
         def fake_run_cmd(cmd, host, user=None):
             if cmd[:2] == ["docker", "inspect"]:
-                inspect_calls.append(cmd)
+                inspect_calls.append((cmd, host, user))
                 return SimpleNamespace(returncode=0, stdout="true\n", stderr="")
             return SimpleNamespace(returncode=0, stdout="", stderr="")
 
@@ -112,7 +113,50 @@ def fake_run_cmd(cmd, host, user=None):
             controller._launch_redis_containers()
 
         self.assertEqual(len(inspect_calls), 1)
-        self.assertIn("canyonos-redis-10-0-0-5", inspect_calls[0])
+        cmd, host, user = inspect_calls[0]
+        self.assertIn("canyonos-redis-localhost", cmd)
+        self.assertEqual(host, "localhost")
+        self.assertIsNone(user)
+
+    def test_local_replica_placements_ignore_per_replica_host_overrides(self):
+        placements = GlobalController._get_replica_placements(
+            {
+                "name": "Workflow",
+                "host": "10.0.0.5",
+                "replicas": [
+                    {"host": "10.0.0.6", "port": 50061},
+                    {"host": "10.0.0.7", "port": 50062},
+                ],
+            }
+        )
+
+        self.assertEqual(placements, [("localhost", 50061), ("localhost", 50062)])
+
+    def test_stop_local_redis_never_uses_ssh_overrides(self):
+        controller = _bare_controller(
+            [
+                {
+                    "name": "Workflow",
+                    "host": "10.0.0.5",
+                    "user": "ubuntu",
+                }
+            ]
+        )
+        controller.redis_containers = {"localhost": "canyonos-redis-localhost"}
+        calls = []
+        controller._run_cmd = lambda cmd, host, user=None: calls.append(
+            (cmd, host, user)
+        )
+
+        controller._stop_redis_containers()
+
+        self.assertEqual(
+            calls,
+            [
+                (["docker", "stop", "canyonos-redis-localhost"], "localhost", None),
+                (["docker", "rm", "canyonos-redis-localhost"], "localhost", None),
+            ],
+        )
 
 
 if __name__ == "__main__":
diff --git a/tests/test_instance_manager_runtime.py b/tests/test_instance_manager_runtime.py
index 5e99caa9..44909680 100644
--- a/tests/test_instance_manager_runtime.py
+++ b/tests/test_instance_manager_runtime.py
@@ -189,6 +189,31 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self):
             ),
         )
 
+    def test_local_instances_ignore_host_and_user_overrides(self):
+        controller = _fake_controller()
+        manager = InstanceManager(controller, controller.redis)
+
+        instance = manager.ensure_instances(
+            [
+                {
+                    "name": "Alpha",
+                    "provider": "local",
+                    "host": "10.0.0.5",
+                    "user": "ubuntu",
+                }
+            ]
+        )[0]
+
+        self.assertEqual(instance["host"], "localhost")
+        self.assertEqual(instance["endpoint"], "localhost:8000")
+        self.assertNotIn("user", instance)
+        self.assertEqual(
+            controller.redis.hgetall("agent_instance:local:Alpha:0")["host"],
+            "localhost",
+        )
+        for call in controller._run_cmd.call_args_list:
+            self.assertEqual(call.args[1:], ("localhost", None))
+
     def test_bootstrap_instance_passes_poll_interval_env_var(self):
         controller = _fake_controller()
         controller.config = {"poll_interval": 7}
@@ -227,17 +252,16 @@ 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):
-            manager.ensure_instances(
-                [
-                    {
-                        "name": "Workflow",
-                        "provider": "local",
-                        "type": "workflow",
-                        "resources": {"cpu": 2, "memory": 1024, "gpu": 1},
-                    }
-                ]
-            )
+        manager.ensure_instances(
+            [
+                {
+                    "name": "Workflow",
+                    "provider": "local",
+                    "type": "workflow",
+                    "resources": {"cpu": 2, "memory": 1024, "gpu": 1},
+                }
+            ]
+        )
 
         self.assertNotIn(
             "-it",
@@ -286,35 +310,6 @@ def test_local_workflow_and_resource_flags_stay_the_same(self):
             ),
         )
 
-    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 self.assertRaises(RuntimeError) as ctx:
-                manager.ensure_instances(
-                    [{"name": "Workflow", "provider": "local", "type": "workflow"}]
-                )
-
-        self.assertIn("api_port 8080", str(ctx.exception))
-        # The orphan-container `docker inspect` probe still runs -- only `docker run` is skipped.
-        run_calls = [
-            c
-            for c in controller._run_cmd.call_args_list
-            if c.args[0][:2] == ["docker", "run"]
-        ]
-        self.assertEqual(run_calls, [])
-
-    def test_plain_agent_bootstrap_ignores_api_port_conflicts(self):
-        """Only `type: workflow` publishes api_port -- a plain agent has nothing to conflict on."""
-        controller = _fake_controller()
-        manager = InstanceManager(controller, controller.redis)
-
-        with patch.object(local_runtime, "_port_bound", return_value=True):
-            manager.ensure_instances([{"name": "Alpha", "provider": "local"}])
-
-        controller._run_cmd.assert_called()
-
     def test_agent_id_is_stable_across_repeated_ensure_instances_calls(self):
         controller = _fake_controller()
         manager = InstanceManager(controller, controller.redis)
diff --git a/tests/test_runtime_ec2.py b/tests/test_runtime_ec2.py
index 21a7f6b2..a205e19e 100644
--- a/tests/test_runtime_ec2.py
+++ b/tests/test_runtime_ec2.py
@@ -1,10 +1,15 @@
 import os
+import stat
+import subprocess
 import sys
 import tempfile
+import threading
 import unittest
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from botocore.exceptions import ClientError
+
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from canyonos_core.controller.cloud_provider_logic.EC2 import _runtime as ec2_runtime
@@ -25,8 +30,13 @@ def __init__(self, public_ip="54.10.20.30", private_ip="10.0.0.30"):
         self.instances = {}
         self.run_requests = []
         self.terminate_requests = []
+        self.import_key_pair_requests = []
         self.waiter = _FakeWaiter()
 
+    def import_key_pair(self, **kwargs):
+        self.import_key_pair_requests.append(kwargs)
+        return {}
+
     def run_instances(self, **kwargs):
         self.run_requests.append(kwargs)
         instance_id = f"i-test{len(self.run_requests)}"
@@ -62,6 +72,11 @@ def setUp(self):
         key_file = tempfile.NamedTemporaryFile(delete=False)
         key_file.close()
         self.key_path = key_file.name
+        os.unlink(self.key_path)
+        subprocess.run(
+            ["ssh-keygen", "-t", "ed25519", "-N", "", "-f", self.key_path, "-q"],
+            check=True,
+        )
         os.chmod(self.key_path, 0o600)
         self.fake_client = _FakeEC2Client()
         self.client_calls = []
@@ -75,7 +90,6 @@ def setUp(self):
                     "region": "us-east-1",
                     "ssh_user": "ubuntu",
                     "ssh_private_key_path": self.key_path,
-                    "public_ip_timeout": 1,
                 },
             },
             registry_url=None,
@@ -94,6 +108,7 @@ def setUp(self):
     def tearDown(self):
         self.client_patch.stop()
         os.unlink(self.key_path)
+        os.unlink(f"{self.key_path}.pub")
         ec2_runtime._controller = self.original_controller
 
     def _make_client(self, service_name, region_name=None):
@@ -104,11 +119,71 @@ def _make_client(self, service_name, region_name=None):
         return self.fake_client
 
     def test_aws_clients_fails_when_required_fields_are_missing(self):
-        self.controller.config["ec2"].pop("ami_id")
+        self.controller.config["ec2"].pop("subnet_id")
 
         with self.assertRaisesRegex(ValueError, "Missing EC2 config"):
             ec2_runtime._aws_clients()
 
+    def test_aws_clients_defaults_ami_id_when_missing(self):
+        self.controller.config["ec2"].pop("ami_id")
+
+        cfg, _ = ec2_runtime._aws_clients()
+
+        self.assertEqual(cfg["ami_id"], ec2_runtime.DEFAULT_AMI_ID)
+
+    def test_aws_clients_defaults_ssh_user_when_missing(self):
+        self.controller.config["ec2"].pop("ssh_user")
+
+        cfg, _ = ec2_runtime._aws_clients()
+
+        self.assertEqual(cfg["ssh_user"], ec2_runtime.DEFAULT_SSH_USER)
+
+    def test_aws_clients_generates_default_key_when_unset(self):
+        self.controller.config["ec2"].pop("ssh_private_key_path")
+        default_dir = tempfile.mkdtemp()
+        default_path = os.path.join(default_dir, "canyonos_ec2")
+
+        try:
+            with patch.object(ec2_runtime, "DEFAULT_SSH_KEY_PATH", default_path):
+                cfg, _ = ec2_runtime._aws_clients()
+                self.assertEqual(
+                    ec2_runtime._ssh_key_path(self.controller.config["ec2"]),
+                    default_path,
+                )
+
+            self.assertTrue(os.path.isfile(default_path))
+            mode = stat.S_IMODE(os.stat(default_path).st_mode)
+            self.assertEqual(mode, 0o600)
+        finally:
+            os.unlink(default_path)
+            if os.path.exists(f"{default_path}.pub"):
+                os.unlink(f"{default_path}.pub")
+            os.rmdir(default_dir)
+
+    def test_default_key_generation_is_race_safe(self):
+        default_dir = tempfile.mkdtemp()
+        default_path = os.path.join(default_dir, "canyonos_ec2")
+
+        try:
+            with patch.object(ec2_runtime, "DEFAULT_SSH_KEY_PATH", default_path):
+                threads = [
+                    threading.Thread(
+                        target=ec2_runtime._generate_default_key, args=(default_path,)
+                    )
+                    for _ in range(5)
+                ]
+                for t in threads:
+                    t.start()
+                for t in threads:
+                    t.join()
+
+            self.assertTrue(os.path.isfile(default_path))
+        finally:
+            os.unlink(default_path)
+            if os.path.exists(f"{default_path}.pub"):
+                os.unlink(f"{default_path}.pub")
+            os.rmdir(default_dir)
+
     def test_aws_clients_rejects_missing_ssh_private_key(self):
         self.controller.config["ec2"]["ssh_private_key_path"] = (
             "/tmp/missing-canyonos-key"
@@ -123,6 +198,40 @@ def test_aws_clients_rejects_insecure_ssh_private_key_permissions(self):
         with self.assertRaisesRegex(ValueError, "insecure permissions 0644"):
             ec2_runtime._aws_clients()
 
+    def test_key_pair_name_scoped_by_project_id_and_stable_for_same_key(self):
+        cfg, _ = ec2_runtime._aws_clients()
+        first_name = cfg["key_pair_name"]
+        self.assertRegex(first_name, r"^canyonos-ec2-default-[0-9a-f]{8}$")
+
+        cfg, _ = ec2_runtime._aws_clients()
+        self.assertEqual(cfg["key_pair_name"], first_name)
+
+        self.controller.config["project_id"] = "my-project"
+        cfg, _ = ec2_runtime._aws_clients()
+        self.assertNotEqual(cfg["key_pair_name"], first_name)
+        self.assertTrue(cfg["key_pair_name"].startswith("canyonos-ec2-my-project-"))
+
+    def test_aws_clients_tolerates_already_imported_key_pair(self):
+        self.fake_client.import_key_pair = MagicMock(
+            side_effect=ClientError(
+                {"Error": {"Code": "InvalidKeyPair.Duplicate", "Message": "boom"}},
+                "ImportKeyPair",
+            )
+        )
+
+        ec2_runtime._aws_clients()
+
+    def test_aws_clients_reraises_other_key_pair_errors(self):
+        self.fake_client.import_key_pair = MagicMock(
+            side_effect=ClientError(
+                {"Error": {"Code": "UnauthorizedOperation", "Message": "boom"}},
+                "ImportKeyPair",
+            )
+        )
+
+        with self.assertRaises(ClientError):
+            ec2_runtime._aws_clients()
+
     def test_provision_uses_ec2_client(self):
         spec = {
             "name": "Tagged",
@@ -135,19 +244,44 @@ def test_provision_uses_ec2_client(self):
 
         request = self.fake_client.run_requests[0]
         self.assertEqual(request["ImageId"], "ami-123456")
-        self.assertNotIn("KeyName", request)
+        self.assertRegex(request["KeyName"], r"^canyonos-ec2-default-[0-9a-f]{8}$")
         self.assertNotIn("UserData", request)
+        self.assertNotIn("IamInstanceProfile", request)
+        self.assertEqual(
+            self.fake_client.import_key_pair_requests[0]["KeyName"],
+            request["KeyName"],
+        )
         self.assertEqual(
             request["TagSpecifications"][0]["Tags"][0],
             {"Key": "Name", "Value": "canyonos-Tagged-2"},
         )
         self.assertEqual(self.fake_client.waiter.calls, [["i-test1"]])
-        self.assertEqual(provisioned["host"], "10.0.0.30")
+        self.assertEqual(provisioned["host"], "54.10.20.30")
         self.assertEqual(
             self.client_calls,
             [{"service_name": "ec2", "region_name": "us-east-1"}],
         )
 
+    def test_provision_falls_back_to_private_ip_when_no_public_ip(self):
+        self.fake_client.public_ip = None
+        spec = {"name": "Tagged", "provider": "EC2", "instance_type": "t3.small"}
+
+        provisioned = ec2_runtime.provision_instance(spec, 0)
+
+        self.assertEqual(provisioned["host"], "10.0.0.30")
+        self.assertIsNone(provisioned["public_host"])
+
+    def test_provision_attaches_instance_profile_when_configured(self):
+        self.controller.config["ec2"]["instance_profile_name"] = "my-custom-profile"
+        spec = {"name": "Tagged", "provider": "EC2", "instance_type": "t3.small"}
+
+        ec2_runtime.provision_instance(spec, 0)
+
+        request = self.fake_client.run_requests[0]
+        self.assertEqual(
+            request["IamInstanceProfile"], {"Name": "my-custom-profile"}
+        )
+
     def test_provision_and_bootstrap_instance_return_runtime_record(self):
         spec = {
             "name": "Tagged",
@@ -165,9 +299,9 @@ def test_provision_and_bootstrap_instance_return_runtime_record(self):
                 provisioned, spec, 2, "agent-id-2"
             )
 
-        self.assertEqual(instance["host"], "10.0.0.30")
-        self.assertEqual(instance["endpoint"], "10.0.0.30:50051")
-        self.assertEqual(instance["redis_host"], "10.0.0.30")
+        self.assertEqual(instance["host"], "54.10.20.30")
+        self.assertEqual(instance["endpoint"], "54.10.20.30:50051")
+        self.assertEqual(instance["redis_host"], "54.10.20.30")
         self.assertEqual(instance["redis_port"], "6390")
         self.assertIn("--i-test1", instance["runtime_id"])