diff --git a/PLAN-cli-lambda.md b/PLAN-cli-lambda.md
index 2d90d12..4f773da 100644
--- a/PLAN-cli-lambda.md
+++ b/PLAN-cli-lambda.md
@@ -24,6 +24,11 @@
 - Documentation authority: this file is authoritative for phase 1. Any older standalone CLI migration spec is advisory only until it is reconciled back into this document.
 - Dependencies: `requests` may be added to runtime dependencies; `responses` may be added to test dependencies.
 - End-of-phase validation: real local CLI invocation against deployed AWS, not just local mocks or container smoke tests.
+- Streaming handlers return lazy event iterables so progress is sent while long-running actions execute and iteration failures become terminal `error` events.
+- Remote requests use separate connect and read timeouts. The connect timeout is 10 seconds, existing commands retain a 30-second read timeout, and long-running actions may opt into a longer read timeout.
+- Phase 3 `launch` uses a 930-second read timeout against a Lambda configured for the 900-second maximum. The workflow polls instance state every 15 seconds for at most 40 attempts and emits progress between polls.
+- The CLI Lambda is pinned to Terraform's `DEVBOX_PARAM_PREFIX`. `launch` rejects a request whose envelope prefix does not match that configured value and never uses a caller-selected prefix for SSM or DynamoDB access.
+- EC2 userdata is read by the local CLI, sent inline, and limited to 16,384 UTF-8 bytes on both sides of the wire.
 - `new` is deferred but remains in scope, even though the current CLI references a missing implementation.
 
 ## Public Contract
@@ -226,7 +231,7 @@ Milestone: `devbox terminate <instance-id-or-project>` runs from the local CLI t
   - [x] transport failure behavior
 - [x] Run `pixi run -e dev python -m pytest` for touched tests.
 - [x] Run `tofu fmt` and `tofu validate`.
-- [ ] Record the local end-to-end termination validation steps and outcome in this file.
+- [x] Record the local end-to-end termination validation steps and outcome in this file.
 
 ### Phase 2 `terminate` Contract
 
@@ -268,38 +273,104 @@ Milestone: `devbox terminate <instance-id-or-project>` runs from the local CLI t
 - `2026-05-29`: Terraform validation passed.
   - Command: `tofu validate`
   - Result: passed
-- End-to-end commands (must be run by DWHS in devbox-deploy/ repo, not for agents)
-  - `tofu plan -out tfplan && tofu apply tfplan` to deploy the updated CLI Lambda and IAM policy
-  - `devbox terminate i-0123456789abcdef0` against the deployed CLI Lambda, expecting the existing success message
-  - `devbox terminate my-project` against the deployed CLI Lambda, expecting project-name resolution to match current behavior
-- Observed outcome: automated validation passed locally; deployed-AWS end-to-end termination validation is still pending operator run.
+- `2026-07-16`: deployed-AWS end-to-end validation was completed by DWHS.
+  - Command: `devbox terminate <instance-id-or-project>` against the deployed CLI Lambda.
+  - Result: the termination command successfully ran through the Lambda and terminated the target instance.
+
+### Phase 2 Handoff Notes
+
+- Automated coverage verifies both supported identifier forms (instance ID and project name), along with not-found, ambiguity, terminal-error, and transport-failure behavior. The deployed-AWS run supplies the required real CLI/Lambda-path validation; another live termination is not required to close this phase.
+- Remaining phase 2 blockers: none. Phase 2 is complete.
+- Next session starts here: begin phase 3 `launch`, starting with the wire contract and HTTP timeout-policy decision.
 
 ## Phase 3: `launch`
 
 Milestone: `devbox launch ...` runs from the local CLI through the deployed CLI Lambda, including userdata handling and current DNS flags.
 
-- [ ] Extend the wire contract for `launch`, including all current CLI options and the inline userdata payload shape.
-- [ ] Refactor launch logic so Lambda can emit structured progress events instead of relying on raw `print` output.
-- [ ] Revisit the CLI HTTP timeout policy before `launch` goes remote. Phase 1 wraps `requests` transport failures cleanly, but still uses a single `timeout=30`; decide whether streamed commands need separate connect/read timeouts or a longer read timeout.
-- [ ] Preserve shared business logic; do not fork a second launch implementation just for the Lambda path.
-- [ ] Keep local preprocessing in the CLI for:
-  - [ ] reading `--userdata-file`
-  - [ ] embedding file contents into the request payload
-  - [ ] rejecting oversized request bodies before transmission if needed
-- [ ] Add the remote `launch` action to the Lambda router.
-- [ ] Expand CLI Lambda IAM only for launch-related operations.
-- [ ] Migrate only `launch` in the CLI to the remote path.
-- [ ] Preserve current flags for DNS behavior and ensure the request contract carries the same semantics.
-- [ ] Add tests for:
-  - [ ] payload construction for all launch options
-  - [ ] userdata inlining
-  - [ ] progress-event rendering
-  - [ ] launch success and failure mapping
-  - [ ] DNS option propagation
-- [ ] Run `pixi run -e dev python -m pytest` for touched tests.
-- [ ] Run `tofu fmt` and `tofu validate`.
+- [x] Extend the wire contract for `launch`, including all current CLI options and the inline userdata payload shape.
+- [x] Refactor launch logic so Lambda can emit structured progress events instead of relying on raw `print` output.
+- [x] Revisit the CLI HTTP timeout policy before `launch` goes remote. Phase 1 wraps `requests` transport failures cleanly, but still uses a single `timeout=30`; decide whether streamed commands need separate connect/read timeouts or a longer read timeout.
+- [x] Preserve shared business logic; do not fork a second launch implementation just for the Lambda path.
+- [x] Keep local preprocessing in the CLI for:
+  - [x] reading `--userdata-file`
+  - [x] embedding file contents into the request payload
+  - [x] rejecting oversized request bodies before transmission if needed
+- [x] Add the remote `launch` action to the Lambda router.
+- [x] Expand CLI Lambda IAM only for launch-related operations.
+- [x] Migrate only `launch` in the CLI to the remote path.
+- [x] Preserve current flags for DNS behavior and ensure the request contract carries the same semantics.
+- [x] Add tests for:
+  - [x] payload construction for all launch options
+  - [x] userdata inlining
+  - [x] progress-event rendering
+  - [x] launch success and failure mapping
+  - [x] DNS option propagation
+- [x] Run `pixi run -e dev python -m pytest` for touched tests.
+- [x] Run `tofu fmt` and `tofu validate`.
 - [ ] Record the local end-to-end launch validation steps and outcome in this file.
 
+### Phase 3 `launch` Contract
+
+- Request payload:
+
+```json
+{
+  "project": "my-project",
+  "instance_type": "t3.medium",
+  "key_pair": "my-key",
+  "volume_size": 100,
+  "base_ami": "ami-0123456789abcdef0",
+  "userdata": "#!/bin/bash\necho ready",
+  "assign_dns": true,
+  "dns_subdomain": "my-project"
+}
+```
+
+- Optional string fields are `null` when omitted. `volume_size` defaults to `0`, and `assign_dns` defaults to `true`.
+- `userdata` contains file contents, never a local path, and may be `null`; non-null content is limited to 16,384 UTF-8 bytes.
+- Success path event sequence:
+  - zero or more `progress` and `warning` events
+  - exactly one `result` event
+  - exactly one terminal `success` event
+- The result contains `project`, `instance_id`, `state`, `instance_type`, `image_id`, `availability_zone`, `private_ip`, `public_ip`, `ssh_username`, and `dns_name`. Fields unavailable from EC2 or optional configuration are `null`.
+
+### Phase 3 Validation Log
+
+- `2026-07-17`: full Python test suite passed.
+  - Command: `pixi run -e dev python -m pytest -q`
+  - Result: `471 passed, 1 warning`
+- `2026-07-17`: Ruff passed for all touched Python files.
+- `2026-07-17`: Terraform formatting check passed.
+  - Command: `tofu fmt -check -recursive`
+- `2026-07-17`: Terraform validation passed.
+  - Command: `tofu validate`
+- `2026-07-28`: launch follow-up restored structured volume-adjustment
+  progress and made best-effort launch-template metadata failures visible as
+  structured warnings, with an application-log fallback outside the workflow.
+  The request and event wire formats are unchanged.
+- `2026-07-28`: focused launch tests passed.
+  - Command: `pixi run -e dev python -m pytest tests/test_launch.py tests/test_launch_workflow.py tests/commands/test_launch_command.py -q`
+  - Result: `93 passed, 1 warning`
+- `2026-07-28`: Ruff passed for the follow-up files.
+  - Command: `pixi run -e dev ruff check src/devbox/launch.py tests/test_launch.py tests/test_launch_workflow.py`
+- `2026-07-28`: full Python test suite passed after the follow-up.
+  - Command: `pixi run -e dev python -m pytest -q`
+  - Result: `479 passed, 1 warning`
+- Deployed-AWS validation to be run by DWHS from the deployment environment:
+  - Deploy the reviewed Terraform plan containing the CLI Lambda image, timeout, environment, and IAM changes.
+  - Run `devbox launch <project> --instance-type <type> --key-pair <key> --userdata-file <file>` and include `--dns-subdomain <label>` when DNS is configured.
+  - Confirm volume-adjustment and instance-state progress events arrive before
+    completion, `devbox status <project>` shows the instance, userdata ran, and
+    the optional DNS name resolves.
+  - Clean up with `devbox terminate <project>`.
+
+### Phase 3 Handoff Notes
+
+- Automated implementation and validation, including the launch progress/error
+  visibility follow-up, are complete. The remaining phase 3 item is an
+  operator-run deployed-AWS acceptance rerun using the flow above.
+- Do not mark phase 3 complete or begin phase 4 until that outcome is recorded here.
+
 ## Phase 4: `new`
 
 Milestone: `devbox new ...` works end-to-end through the deployed CLI Lambda and is no longer dependent on a missing local implementation.
diff --git a/main.tf b/main.tf
index 5b092c8..357b5b6 100644
--- a/main.tf
+++ b/main.tf
@@ -29,8 +29,11 @@ module "snapshot_lambda" {
 module "cli_lambda" {
   source = "./modules/cli-lambda"
 
-  prefix       = var.prefix
-  param_prefix = "/${var.prefix}"
+  prefix                = var.prefix
+  param_prefix          = "/${var.prefix}"
+  snapshot_table_arn    = module.snapshot_lambda.dynamodb_table_arn
+  ec2_instance_role_arn = module.devbox.ec2_role_arn
+  dns_provider          = var.dns_provider
 }
 
 module "dns_cleanup_lambda" {
diff --git a/modules/cli-lambda/main.tf b/modules/cli-lambda/main.tf
index e15c13b..1fcc50f 100644
--- a/modules/cli-lambda/main.tf
+++ b/modules/cli-lambda/main.tf
@@ -13,7 +13,7 @@ locals {
     for relpath in sort(local.lambda_source_files) : filesha256("${local.repo_root}/${relpath}")
   ]))
   # Keep one statement per CLI action with the full permission set that action needs.
-  command_policy_statements = [
+  command_policy_statements = concat([
     {
       sid = "StatusCmdPermissions"
       actions = [
@@ -31,8 +31,52 @@ locals {
         "ec2:TerminateInstances"
       ]
       resources = ["*"]
+    },
+    {
+      sid = "LaunchCmdEc2Permissions"
+      actions = [
+        "ec2:CreateTags",
+        "ec2:DescribeImages",
+        "ec2:DescribeInstances",
+        "ec2:DescribeLaunchTemplates",
+        "ec2:DescribeLaunchTemplateVersions",
+        "ec2:DescribeSubnets",
+        "ec2:RunInstances"
+      ]
+      resources = ["*"]
+    },
+    {
+      sid       = "LaunchCmdSsmPermissions"
+      actions   = ["ssm:GetParameter"]
+      resources = ["arn:aws:ssm:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:parameter/${trim(var.param_prefix, "/")}/*"]
+    },
+    {
+      sid = "LaunchCmdDynamoDbPermissions"
+      actions = [
+        "dynamodb:GetItem",
+        "dynamodb:PutItem",
+        "dynamodb:UpdateItem"
+      ]
+      resources = [var.snapshot_table_arn]
+    },
+    {
+      sid       = "LaunchCmdPassRolePermissions"
+      actions   = ["iam:PassRole"]
+      resources = [var.ec2_instance_role_arn]
     }
-  ]
+    ],
+    var.dns_provider == "route53" ? [
+      {
+        sid = "LaunchCmdRoute53Permissions"
+        actions = [
+          "route53:ChangeResourceRecordSets",
+          "route53:ListHostedZonesByName",
+          "route53:ListResourceRecordSets"
+        ]
+        resources = ["*"]
+      }
+    ] : []
+  )
 }
 
 resource "aws_ecr_repository" "cli" {
@@ -126,13 +170,14 @@ resource "aws_lambda_function" "cli" {
   package_type  = "Image"
   image_uri     = local.image_uri
   role          = aws_iam_role.lambda_role.arn
-  timeout       = 30
+  timeout       = 900
   memory_size   = 256
 
   environment {
     variables = {
       AWS_LWA_INVOKE_MODE          = "response_stream"
       AWS_LWA_READINESS_CHECK_PATH = "/healthz"
+      DEVBOX_PARAM_PREFIX          = var.param_prefix
       PORT                         = "8080"
     }
   }
diff --git a/modules/cli-lambda/variables.tf b/modules/cli-lambda/variables.tf
index 07f297a..e70b6b9 100644
--- a/modules/cli-lambda/variables.tf
+++ b/modules/cli-lambda/variables.tf
@@ -9,3 +9,24 @@ variable "param_prefix" {
   type        = string
   default     = "/devbox"
 }
+
+variable "snapshot_table_arn" {
+  description = "ARN of the main DevBox snapshot table"
+  type        = string
+}
+
+variable "ec2_instance_role_arn" {
+  description = "ARN of the IAM role attached to launched DevBox instances"
+  type        = string
+}
+
+variable "dns_provider" {
+  description = "DNS provider used for optional launch-time CNAME assignment"
+  type        = string
+  default     = "none"
+
+  validation {
+    condition     = contains(["none", "cloudflare", "route53"], var.dns_provider)
+    error_message = "dns_provider must be one of: none, cloudflare, route53."
+  }
+}
diff --git a/modules/devbox/outputs.tf b/modules/devbox/outputs.tf
index 036f869..758669d 100644
--- a/modules/devbox/outputs.tf
+++ b/modules/devbox/outputs.tf
@@ -7,3 +7,8 @@ output "launch_template_names" {
   description = "Names of the Devbox launch templates"
   value       = aws_launch_template.base[*].name
 }
+
+output "ec2_role_arn" {
+  description = "ARN of the IAM role passed to launched DevBox instances"
+  value       = aws_iam_role.ec2_role.arn
+}
diff --git a/src/devbox/__init__.py b/src/devbox/__init__.py
index a5e60a8..807f083 100644
--- a/src/devbox/__init__.py
+++ b/src/devbox/__init__.py
@@ -10,7 +10,6 @@
 # Import key components for easier access
 from .devbox_manager import DevBoxManager
 from .console_output import ConsoleOutput
-from .launch import main as launch_instance
 from .cli import cli as devbox_cli
 
 # Alias for backward compatibility
@@ -19,7 +18,6 @@
 __all__ = [
     "DevBoxManager",
     "ConsoleOutput",
-    "launch_instance",
     "devbox_cli",
     "main",
 ]
diff --git a/src/devbox/cli.py b/src/devbox/cli.py
index 9afe712..795403b 100644
--- a/src/devbox/cli.py
+++ b/src/devbox/cli.py
@@ -10,6 +10,7 @@
 from .remote_client import normalize_param_prefix
 from .commands.status import run_status_command
 from .commands.terminate import run_terminate_command
+from .commands.launch import run_launch_command
 from .devbox_manager import DevBoxManager
 from .console_output import ConsoleOutput
 
@@ -179,12 +180,10 @@ def launch(
     PROJECT is the name of the project to launch.
     Optionally pass cloud-init user data with --userdata-file.
     """
-    from .launch import launch_programmatic
-
     console = ctx.obj["console"]
 
     try:
-        launch_programmatic(
+        run_launch_command(
             project=project,
             instance_type=instance_type,
             key_pair=key_pair,
@@ -194,6 +193,7 @@ def launch(
             userdata_file=userdata_file,
             assign_dns=not no_assign_dns,
             dns_subdomain=dns_subdomain,
+            console=console,
         )
     except Exception as e:
         console.print_error(f"Failed to launch instance: {str(e)}")
diff --git a/src/devbox/cli_lambda/app.py b/src/devbox/cli_lambda/app.py
index 0e4d314..61ae8e0 100644
--- a/src/devbox/cli_lambda/app.py
+++ b/src/devbox/cli_lambda/app.py
@@ -11,6 +11,7 @@
 
 from ..commands.status import handle_status_action
 from ..commands.terminate import handle_terminate_action
+from ..commands.launch import handle_launch_action
 from ..cli_protocol import NDJSON_MIME_TYPE, CliAction, CliEventType
 from .contracts import (
     CliRequestEnvelope,
@@ -20,15 +21,16 @@
     parse_request_envelope,
 )
 
-ActionHandler = Callable[[CliRequestEnvelope], list[dict[str, Any]]]
+ActionHandler = Callable[[CliRequestEnvelope], Iterable[dict[str, Any]]]
 
 ACTION_HANDLERS: dict[CliAction, ActionHandler] = {
     CliAction.STATUS: handle_status_action,
     CliAction.TERMINATE: handle_terminate_action,
+    CliAction.LAUNCH: handle_launch_action,
 }
 
 
-def dispatch_action(envelope: CliRequestEnvelope) -> list[dict[str, object]]:
+def dispatch_action(envelope: CliRequestEnvelope) -> Iterable[dict[str, Any]]:
     """Dispatch one validated CLI Lambda request.
 
     Parameters
@@ -53,7 +55,7 @@ def dispatch_action(envelope: CliRequestEnvelope) -> list[dict[str, object]]:
     return handler(envelope)
 
 
-def execute_action(envelope: CliRequestEnvelope) -> list[dict[str, object]]:
+def execute_action(envelope: CliRequestEnvelope) -> Iterable[dict[str, Any]]:
     """Execute one CLI Lambda request.
 
     Parameters
@@ -74,11 +76,11 @@ def execute_action(envelope: CliRequestEnvelope) -> list[dict[str, object]]:
         400 instead of an application ``error`` event.
     """
     try:
-        return dispatch_action(envelope)
+        yield from dispatch_action(envelope)
     except InvalidCliRequestError:
         raise
     except Exception as exc:
-        return [build_event(CliEventType.ERROR, envelope.action, str(exc))]
+        yield build_event(CliEventType.ERROR, envelope.action, str(exc))
 
 
 def stream_events(events: Iterable[dict[str, object]]) -> StreamingResponse:
diff --git a/src/devbox/cli_protocol.py b/src/devbox/cli_protocol.py
index f6f180d..75d03c4 100644
--- a/src/devbox/cli_protocol.py
+++ b/src/devbox/cli_protocol.py
@@ -14,6 +14,7 @@ class CliAction(StrEnum):
 
     STATUS = "status"
     TERMINATE = "terminate"
+    LAUNCH = "launch"
 
 
 class CliEventType(StrEnum):
diff --git a/src/devbox/commands/launch.py b/src/devbox/commands/launch.py
new file mode 100644
index 0000000..1d749d1
--- /dev/null
+++ b/src/devbox/commands/launch.py
@@ -0,0 +1,167 @@
+"""Launch command helpers shared by the CLI and CLI Lambda."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any, Iterator, Optional
+
+from ..cli_lambda.contracts import CliRequestEnvelope, build_event
+from ..cli_protocol import CliAction, CliEventType
+from ..launch import LaunchRequest, LaunchResult, LaunchUpdate, iter_launch_workflow
+from ..remote_client import invoke_action, normalize_param_prefix
+
+MAX_USERDATA_BYTES = 16_384
+LAUNCH_READ_TIMEOUT_SECONDS = 930
+
+
+def validate_userdata(userdata: Optional[str]) -> Optional[str]:
+    """Validate optional raw EC2 userdata and return it unchanged."""
+    if userdata is None:
+        return None
+    if not isinstance(userdata, str):
+        raise ValueError("The `launch` payload field `userdata` must be a string or null.")
+    if len(userdata.encode("utf-8")) > MAX_USERDATA_BYTES:
+        raise ValueError(
+            f"EC2 userdata must not exceed {MAX_USERDATA_BYTES} UTF-8 bytes."
+        )
+    return userdata
+
+
+def read_userdata_file(userdata_file: Optional[str]) -> Optional[str]:
+    """Read and validate a local userdata file for request inlining."""
+    if userdata_file is None:
+        return None
+    return validate_userdata(Path(userdata_file).read_text(encoding="utf-8"))
+
+
+def build_launch_payload(
+    *,
+    project: str,
+    instance_type: Optional[str],
+    key_pair: Optional[str],
+    volume_size: int,
+    base_ami: Optional[str],
+    userdata: Optional[str],
+    assign_dns: bool,
+    dns_subdomain: Optional[str],
+) -> dict[str, Any]:
+    """Build the wire payload for a remote launch request."""
+    return {
+        "project": project,
+        "instance_type": instance_type,
+        "key_pair": key_pair,
+        "volume_size": volume_size,
+        "base_ami": base_ami,
+        "userdata": validate_userdata(userdata),
+        "assign_dns": assign_dns,
+        "dns_subdomain": dns_subdomain,
+    }
+
+
+def run_launch_command(
+    *,
+    project: str,
+    instance_type: Optional[str],
+    key_pair: Optional[str],
+    volume_size: int,
+    base_ami: Optional[str],
+    param_prefix: str,
+    userdata_file: Optional[str],
+    assign_dns: bool,
+    dns_subdomain: Optional[str],
+    console: Any,
+) -> None:
+    """Invoke the remote launch action and render its final result."""
+    userdata = read_userdata_file(userdata_file)
+    result = invoke_action(
+        action=CliAction.LAUNCH,
+        payload=build_launch_payload(
+            project=project,
+            instance_type=instance_type,
+            key_pair=key_pair,
+            volume_size=volume_size,
+            base_ami=base_ami,
+            userdata=userdata,
+            assign_dns=assign_dns,
+            dns_subdomain=dns_subdomain,
+        ),
+        param_prefix=param_prefix,
+        console=console,
+        read_timeout=LAUNCH_READ_TIMEOUT_SECONDS,
+    )
+    console.print_launch_result(result)
+
+
+def _optional_string(payload: dict[str, Any], name: str) -> Optional[str]:
+    value = payload.get(name)
+    if value is not None and (not isinstance(value, str) or not value.strip()):
+        raise ValueError(f"The `launch` payload field `{name}` must be a string or null.")
+    return value
+
+
+def validate_launch_payload(payload: dict[str, Any], param_prefix: str) -> LaunchRequest:
+    """Validate a launch payload and return normalized workflow inputs."""
+    project = payload.get("project")
+    if not isinstance(project, str) or not project.replace("-", "").isalnum():
+        raise ValueError(
+            "The `launch` payload field `project` must be alphanumeric with optional hyphens."
+        )
+
+    volume_size = payload.get("volume_size", 0)
+    if isinstance(volume_size, bool) or not isinstance(volume_size, int) or volume_size < 0:
+        raise ValueError(
+            "The `launch` payload field `volume_size` must be a non-negative integer."
+        )
+    assign_dns = payload.get("assign_dns", True)
+    if not isinstance(assign_dns, bool):
+        raise ValueError("The `launch` payload field `assign_dns` must be a boolean.")
+
+    return LaunchRequest(
+        project=project,
+        instance_type=_optional_string(payload, "instance_type"),
+        key_pair=_optional_string(payload, "key_pair"),
+        volume_size=volume_size,
+        base_ami=_optional_string(payload, "base_ami"),
+        param_prefix=param_prefix,
+        userdata=validate_userdata(payload.get("userdata")),
+        assign_dns=assign_dns,
+        dns_subdomain=_optional_string(payload, "dns_subdomain"),
+    )
+
+
+def configured_param_prefix(envelope_prefix: str) -> str:
+    """Return the configured Lambda prefix after rejecting request mismatches."""
+    configured = os.environ.get("DEVBOX_PARAM_PREFIX")
+    if not configured:
+        raise RuntimeError("CLI Lambda is missing DEVBOX_PARAM_PREFIX configuration.")
+    normalized_configured = normalize_param_prefix(configured)
+    if normalize_param_prefix(envelope_prefix) != normalized_configured:
+        raise ValueError("Request parameter prefix does not match the CLI Lambda configuration.")
+    return normalized_configured
+
+
+def handle_launch_action(envelope: CliRequestEnvelope) -> Iterator[dict[str, Any]]:
+    """Execute launch and lazily map workflow updates to NDJSON events."""
+    param_prefix = configured_param_prefix(envelope.param_prefix)
+    launch_request = validate_launch_payload(envelope.payload, param_prefix)
+    saw_result = False
+    for update in iter_launch_workflow(launch_request):
+        if isinstance(update, LaunchUpdate):
+            event_type = (
+                CliEventType.WARNING
+                if update.kind == "warning"
+                else CliEventType.PROGRESS
+            )
+            yield build_event(event_type, CliAction.LAUNCH, update.message)
+        elif isinstance(update, LaunchResult):
+            saw_result = True
+            yield build_event(
+                CliEventType.RESULT,
+                CliAction.LAUNCH,
+                "Instance launched",
+                update.to_dict(),
+            )
+    if not saw_result:
+        raise RuntimeError("Launch workflow completed without a result.")
+    yield build_event(CliEventType.SUCCESS, CliAction.LAUNCH, "Launch complete")
diff --git a/src/devbox/console_output.py b/src/devbox/console_output.py
index 131954a..9c2c270 100644
--- a/src/devbox/console_output.py
+++ b/src/devbox/console_output.py
@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import List, Dict, Any, Optional, Union
+from typing import Any, Dict, List
 
 from rich.table import Table
 from rich.console import Console
@@ -166,6 +166,40 @@ def print_warning(self, message: str) -> None:
             message: Warning message to display
         """
         self.console.print(f"[yellow]Warning: {message}[/yellow]")
+
+    def print_progress(self, message: str) -> None:
+        """Print a progress message.
+
+        Parameters
+        ----------
+        message : str
+            Progress message to display.
+        """
+        self.console.print(f"[blue]{message}[/blue]")
+
+    def print_launch_result(self, result: dict) -> None:
+        """Render the final details for a launched instance."""
+        self.console.print("\n[bold green]Instance Launched Successfully[/bold green]")
+        labels = (
+            ("Instance ID", result.get("instance_id")),
+            ("State", result.get("state")),
+            ("Type", result.get("instance_type")),
+            ("AMI", result.get("image_id")),
+            ("Availability Zone", result.get("availability_zone")),
+            ("Private IP", result.get("private_ip")),
+            ("Public IP", result.get("public_ip")),
+            ("DNS", result.get("dns_name")),
+        )
+        for label, value in labels:
+            if value:
+                self.console.print(f"{label}: {value}")
+        public_ip = result.get("public_ip")
+        if public_ip:
+            username = result.get("ssh_username") or "<username>"
+            self.console.print(
+                f"\nYou can SSH into the instance using:\n"
+                f"ssh -i /path/to/your-key.pem {username}@{public_ip}"
+            )
     
     @staticmethod
     def _format_timedelta(delta) -> str:
diff --git a/src/devbox/launch.py b/src/devbox/launch.py
index fe13c9d..1e6f6b2 100644
--- a/src/devbox/launch.py
+++ b/src/devbox/launch.py
@@ -4,8 +4,11 @@
 from __future__ import annotations
 
 import argparse
+import logging
 import sys
-from typing import Dict, Any, Optional, Tuple, List, TYPE_CHECKING
+import time
+from dataclasses import asdict, dataclass
+from typing import Any, Callable, Dict, Iterator, List, Literal, Optional, TYPE_CHECKING, Tuple
 
 from botocore.exceptions import ClientError
 
@@ -34,6 +37,8 @@
 EC2Client = Any
 EC2ServiceResource = Any
 
+LOGGER = logging.getLogger(__name__)
+
 
 def read_userdata_file(filepath: Optional[str]) -> Optional[str]:
     """Read userdata file and return its content.
@@ -148,7 +153,11 @@ def get_project_snapshot(
 
 
 def get_volume_info(
-    ec2: Any, image_id: str, min_volume_size: int = 0
+    ec2: Any,
+    image_id: str,
+    min_volume_size: int = 0,
+    *,
+    on_progress: Optional[Callable[[str], None]] = None,
 ) -> Tuple[List[Dict[str, Any]], int]:
     """Get and validate volume information for an AMI.
 
@@ -156,6 +165,7 @@ def get_volume_info(
         ec2: EC2 client
         image_id: ID of the AMI to get volume info for
         min_volume_size: Minimum size (GiB) for the largest volume
+        on_progress: Optional callback for volume-adjustment progress
 
     Returns:
         Tuple of (volumes, largest_volume_size) where:
@@ -193,16 +203,17 @@ def get_volume_info(
     # Update volumes if needed
     if min_volume_size > 0 and largest_volume_size < min_volume_size:
         if largest_volume:
-            print(
-                f"Increasing volume size from {largest_volume_size} GiB to {min_volume_size} GiB"
-            )
+            if on_progress is not None:
+                on_progress(
+                    f"Increasing volume size from {largest_volume_size} GiB "
+                    f"to {min_volume_size} GiB"
+                )
             largest_volume["Ebs"]["VolumeSize"] = min_volume_size
             largest_volume_size = min_volume_size
             # Ensure volume type is set to a modern type
             if "VolumeType" not in largest_volume["Ebs"]:
                 largest_volume["Ebs"]["VolumeType"] = "gp3"
         else:
-            print(f"Creating new volume of size {min_volume_size} GiB")
             volumes.append(
                 {
                     "DeviceName": "/dev/sda1",
@@ -214,17 +225,43 @@ def get_volume_info(
                     },
                 }
             )
+            if on_progress is not None:
+                on_progress(f"Creating new volume of size {min_volume_size} GiB")
             largest_volume_size = min_volume_size
 
-    return volumes, largest_volume_size if largest_volume else 0
+    return volumes, largest_volume_size
+
+
+def _report_launch_template_warning(
+    message: str, on_warning: Optional[Callable[[str], None]]
+) -> None:
+    """Send a best-effort metadata warning to the workflow or application log."""
+    if on_warning is not None:
+        on_warning(message)
+    else:
+        LOGGER.warning(message)
+
 
+def _client_error_summary(error: ClientError) -> str:
+    """Return the AWS error code and message from a client error."""
+    details = error.response.get("Error", {})
+    code = details.get("Code", "Unknown")
+    message = details.get("Message", str(error))
+    return f"{code}: {message}"
 
-def get_launch_template_info(ec2: Any, lt_ids: List[str]) -> Dict[str, Dict[str, str]]:
+
+def get_launch_template_info(
+    ec2: Any,
+    lt_ids: List[str],
+    *,
+    on_warning: Optional[Callable[[str], None]] = None,
+) -> Dict[str, Dict[str, str]]:
     """Get availability zone information for launch templates.
 
     Args:
         ec2: EC2 client
         lt_ids: List of launch template IDs
+        on_warning: Optional callback for non-fatal metadata lookup failures
 
     Returns:
         Dictionary mapping launch template IDs to AZ info with keys:
@@ -286,11 +323,21 @@ def get_launch_template_info(ec2: Any, lt_ids: List[str]) -> Dict[str, Dict[str,
                             az_name = subnet_desc["Subnets"][0].get(
                                 "AvailabilityZone", az_name
                             )
-                    except ClientError:
-                        pass
+                    except ClientError as error:
+                        _report_launch_template_warning(
+                            f"Unable to inspect subnet {subnet_id} for launch "
+                            f"template {lt_id}: {_client_error_summary(error)}; "
+                            f"using availability-zone label {az_name}",
+                            on_warning,
+                        )
 
-        except ClientError as e:
-            print(f"Warning: Error processing launch template {lt_id}: {e}")
+        except ClientError as error:
+            _report_launch_template_warning(
+                f"Unable to inspect launch template {lt_id}: "
+                f"{_client_error_summary(error)}; using fallback "
+                f"availability-zone label {az_name}",
+                on_warning,
+            )
 
         az_info[lt_id] = {"name": az_name, "index": az_index}
 
@@ -330,8 +377,6 @@ def launch_instance(
         - error: Exception if an error occurred, None otherwise
     """
     try:
-        print(f"Attempting to launch instance in {az_name}...")
-
         # Prepare common tags for all resources
         common_tags = [
             {"Key": "Name", "Value": f"devbox-{project}"},
@@ -380,18 +425,11 @@ def launch_instance(
         instance_id = instance_dct["InstanceId"]
         instance = ec2_resource.Instance(instance_id)
 
-        print(
-            f"Instance launched in {az_name}: {instance_id}. Waiting for running state..."
-        )
         return instance, instance_id, None
 
     except ClientError as e:
-        error_code = e.response.get("Error", {}).get("Code", "UnknownError")
-        error_message = e.response.get("Error", {}).get("Message", str(e))
-        print(f"Failed to launch in {az_name}: {error_code} - {error_message}")
         return None, None, e
     except Exception as e:
-        print(f"Unexpected error launching instance: {str(e)}")
         return None, None, e
 
 
@@ -558,10 +596,7 @@ def update_instance_status(
                 f"Unexpected status: {status} (expected 'nonexistent', 'READY', or 'LAUNCHING')"
             )
 
-    except ClientError as e:
-        error_code = e.response.get("Error", {}).get("Code", "UnknownError")
-        error_msg = f"Error updating DynamoDB for {project}: {error_code} - {str(e)}"
-        print(error_msg)
+    except ClientError:
         raise
 
 
@@ -741,9 +776,6 @@ def determine_ami(item: Dict[str, Any], base_ami: Optional[str]) -> str:
             "to create a new snapshot."
         )
 
-    if restored_ami and base_ami:
-        print("Warning: base AMI is ignored when restoring from existing snapshot")
-
     return ami_to_use
 
 
@@ -900,6 +932,285 @@ def display_instance_info(ec2: Any, instance_id: str, project: str, table: Any)
         print(f"\nWarning: Could not get instance details: {str(e)}")
 
 
+@dataclass(frozen=True)
+class LaunchRequest:
+    """Validated inputs for one DevBox launch workflow."""
+
+    project: str
+    instance_type: Optional[str] = None
+    key_pair: Optional[str] = None
+    volume_size: int = 0
+    base_ami: Optional[str] = None
+    param_prefix: str = "/devbox"
+    userdata: Optional[str] = None
+    assign_dns: bool = True
+    dns_subdomain: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class LaunchUpdate:
+    """One progress or warning update produced during launch."""
+
+    kind: Literal["progress", "warning"]
+    message: str
+
+
+@dataclass(frozen=True)
+class LaunchResult:
+    """Serializable details for a successfully launched instance."""
+
+    project: str
+    instance_id: str
+    state: str
+    instance_type: str
+    image_id: str
+    availability_zone: Optional[str]
+    private_ip: Optional[str]
+    public_ip: Optional[str]
+    ssh_username: Optional[str]
+    dns_name: Optional[str]
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return the result as a wire-ready mapping."""
+        return asdict(self)
+
+
+def _instance_state(instance: Any) -> str:
+    """Return the current state name from an EC2 resource."""
+    state = getattr(instance, "state", None)
+    if isinstance(state, dict):
+        return str(state.get("Name", "unknown"))
+    data = getattr(getattr(instance, "meta", None), "data", {}) or {}
+    return str(data.get("State", {}).get("Name", "unknown"))
+
+
+def _resolve_ssh_username(
+    table: Any,
+    project: str,
+    image_id: str,
+    ec2: Any,
+) -> Tuple[Optional[str], Optional[str]]:
+    """Resolve and persist the SSH username, returning a warning if unavailable."""
+    try:
+        item = table.get_item(Key={"project": project}).get("Item", {})
+        username = str(item.get("Username", "")).strip()
+        if username:
+            return username, None
+
+        image = utils.get_image(image_id, ec2_client=ec2)
+        if image is None:
+            return None, f"Could not determine SSH username: AMI {image_id} not found"
+        username = determine_ssh_username(
+            image.get("Name", ""),
+            image.get("Description", ""),
+        )
+        if not username:
+            return None, "Could not determine SSH username from AMI metadata"
+        table.update_item(
+            Key={"project": project},
+            UpdateExpression="SET Username = :u",
+            ExpressionAttributeValues={":u": username},
+        )
+        return username, None
+    except Exception as exc:
+        return None, f"Could not determine SSH username: {exc}"
+
+
+def iter_launch_workflow(
+    request: LaunchRequest,
+    *,
+    aws: Optional[Dict[str, Any]] = None,
+    sleeper: Callable[[float], None] = time.sleep,
+    poll_interval_seconds: float = 15,
+    max_poll_attempts: int = 40,
+) -> Iterator[LaunchUpdate | LaunchResult]:
+    """Launch a DevBox and yield structured updates followed by one result."""
+    if not request.project.replace("-", "").isalnum():
+        raise ValueError("Project name must be alphanumeric with optional hyphens")
+    if request.volume_size < 0:
+        raise ValueError("Volume size must be a non-negative number")
+
+    active_aws = aws or initialize_aws_clients()
+    yield LaunchUpdate("progress", f"Loading launch configuration for {request.project}")
+    config = get_launch_config(active_aws, request.param_prefix, request.project)
+    status = validate_project_status(config["item"], request.project)
+
+    instance_type = request.instance_type or config["item"].get("LastInstanceType")
+    if not instance_type:
+        raise ValueError(
+            "No instance type specified and no previous instance type found for this project"
+        )
+    yield LaunchUpdate("progress", f"Using instance type: {instance_type}")
+
+    key_pair = request.key_pair or config["item"].get("LastKeyPair")
+    if not key_pair:
+        raise ValueError(
+            "No key pair specified and no previous keypair found for this project"
+        )
+    yield LaunchUpdate("progress", f"Using keypair: {key_pair}")
+
+    image_id = determine_ami(config["item"], request.base_ami)
+    if config["item"].get("RestoreAmi") and request.base_ami:
+        yield LaunchUpdate(
+            "warning", "Base AMI is ignored when restoring from existing snapshot"
+        )
+    yield LaunchUpdate("progress", f"Using AMI: {image_id}")
+
+    volume_progress: List[str] = []
+    volumes, _ = get_volume_info(
+        active_aws["ec2"],
+        image_id,
+        request.volume_size,
+        on_progress=volume_progress.append,
+    )
+    for message in volume_progress:
+        yield LaunchUpdate("progress", message)
+
+    template_warnings: List[str] = []
+    az_info = get_launch_template_info(
+        active_aws["ec2"],
+        config["lt_ids"],
+        on_warning=template_warnings.append,
+    )
+    for message in template_warnings:
+        yield LaunchUpdate("warning", message)
+
+    instance = None
+    instance_id = None
+    instance_info: Dict[str, Any] = {}
+    last_error: Optional[Exception] = None
+    for launch_template_id in config["lt_ids"]:
+        az_name = az_info[launch_template_id]["name"]
+        yield LaunchUpdate("progress", f"Attempting to launch instance in {az_name}")
+        candidate, candidate_id, error = launch_instance(
+            ec2=active_aws["ec2"],
+            ec2_resource=active_aws["ec2_resource"],
+            launch_template_id=launch_template_id,
+            image_id=image_id,
+            instance_type=instance_type,
+            key_name=key_pair,
+            volumes=volumes,
+            project=request.project,
+            az_name=az_name,
+            userdata=request.userdata,
+        )
+        if candidate is not None and candidate_id is not None:
+            instance = candidate
+            instance_id = candidate_id
+            instance_info = candidate.meta.data
+            break
+        last_error = error
+        yield LaunchUpdate("warning", f"Launch attempt in {az_name} failed: {error}")
+
+    if instance is None or instance_id is None:
+        message = "Failed to launch instance in all availability zones"
+        if last_error is not None:
+            message += f": {last_error}"
+        raise RuntimeError(message)
+
+    yield LaunchUpdate(
+        "progress", f"Instance {instance_id} launched; waiting for running state"
+    )
+    terminal_states = {"shutting-down", "terminated", "stopping", "stopped"}
+    for attempt in range(max_poll_attempts):
+        try:
+            instance.reload()
+            state = _instance_state(instance)
+        except ClientError as exc:
+            error_code = exc.response.get("Error", {}).get("Code", "")
+            if error_code != "InvalidInstanceID.NotFound":
+                raise
+            state = "not yet visible to EC2"
+        if state == "running":
+            instance_info = instance.meta.data
+            break
+        if state in terminal_states:
+            raise RuntimeError(
+                f"Instance {instance_id} entered terminal state {state} while launching"
+            )
+        if attempt == max_poll_attempts - 1:
+            raise TimeoutError(
+                f"Instance {instance_id} did not reach running state after "
+                f"{max_poll_attempts} checks"
+            )
+        yield LaunchUpdate(
+            "progress", f"Instance {instance_id} is {state}; waiting for running state"
+        )
+        sleeper(poll_interval_seconds)
+
+    cname_domain = config["item"].get("CNAMEDomain")
+    dns_name: Optional[str] = None
+    if request.assign_dns:
+        try:
+            public_dns_name = str(getattr(instance, "public_dns_name", "") or "")
+            if not public_dns_name:
+                yield LaunchUpdate(
+                    "warning", "Instance does not have a public DNS name; skipping DNS assignment"
+                )
+            else:
+                dns_manager = DNSManager.from_ssm(
+                    param_prefix=request.param_prefix,
+                    ssm_client=active_aws["ssm"],
+                )
+                if dns_manager.provider is None:
+                    yield LaunchUpdate("progress", "DNS is not configured; skipping assignment")
+                else:
+                    preferred = request.dns_subdomain
+                    if preferred is None and cname_domain:
+                        preferred = dns_manager.normalize_subdomain(cname_domain)
+                        if preferred is None:
+                            yield LaunchUpdate(
+                                "warning",
+                                f"Stored DNS value '{cname_domain}' is invalid; using project name",
+                            )
+                    resolved = dns_manager.sanitize_dns_name(preferred or request.project)
+                    dns_name = dns_manager.assign_cname(
+                        subdomain=resolved,
+                        instance_public_dns=public_dns_name,
+                    )
+                    if dns_name:
+                        cname_domain = resolved
+                        yield LaunchUpdate("progress", f"Ensured DNS CNAME: {dns_name}")
+        except Exception as exc:
+            yield LaunchUpdate("warning", f"DNS assignment failed: {exc}")
+    elif cname_domain:
+        yield LaunchUpdate("progress", "DNS assignment disabled; existing DNS entry preserved")
+    else:
+        yield LaunchUpdate("progress", "DNS assignment disabled; no DNS record created")
+
+    ssh_username, username_warning = _resolve_ssh_username(
+        config["table"], request.project, image_id, active_aws["ec2"]
+    )
+    if username_warning:
+        yield LaunchUpdate("warning", username_warning)
+
+    update_instance_status(
+        table=config["table"],
+        project=request.project,
+        status=status,
+        instance_id=instance_id,
+        image_id=image_id,
+        instance_type=instance_type,
+        key_pair=key_pair,
+        instance_info=instance_info,
+        cname_domain=cname_domain,
+    )
+
+    placement = instance_info.get("Placement", {})
+    yield LaunchResult(
+        project=request.project,
+        instance_id=instance_id,
+        state=instance_info.get("State", {}).get("Name", _instance_state(instance)),
+        instance_type=instance_info.get("InstanceType", instance_type),
+        image_id=instance_info.get("ImageId", image_id),
+        availability_zone=placement.get("AvailabilityZone"),
+        private_ip=instance_info.get("PrivateIpAddress"),
+        public_ip=instance_info.get("PublicIpAddress"),
+        ssh_username=ssh_username,
+        dns_name=dns_name,
+    )
+
+
 def launch_programmatic(
     project: str,
     instance_type: Optional[str] = None,
diff --git a/src/devbox/remote_client.py b/src/devbox/remote_client.py
index df6df39..e256ee8 100644
--- a/src/devbox/remote_client.py
+++ b/src/devbox/remote_client.py
@@ -29,6 +29,10 @@ class RemoteInvocationError(utils.DevBoxError):
     """Error raised when a remote CLI Lambda invocation fails."""
 
 
+DEFAULT_CONNECT_TIMEOUT_SECONDS = 10
+DEFAULT_READ_TIMEOUT_SECONDS = 30
+
+
 def normalize_param_prefix(param_prefix: str) -> str:
     """Normalize a parameter prefix for CLI Lambda discovery.
 
@@ -270,6 +274,7 @@ def invoke_action(
     payload: dict[str, Any],
     param_prefix: str,
     console: Any | None = None,
+    read_timeout: float = DEFAULT_READ_TIMEOUT_SECONDS,
 ) -> dict[str, Any]:
     """Invoke a CLI Lambda action and return the result payload.
 
@@ -283,6 +288,8 @@ def invoke_action(
         Parameter prefix used to discover the Function URL.
     console : Any | None, optional
         Console-like object used to surface streamed warnings.
+    read_timeout : float, optional
+        Maximum seconds to wait between response bytes after connecting.
 
     Returns
     -------
@@ -307,7 +314,7 @@ def invoke_action(
             data=body,
             headers=headers,
             stream=True,
-            timeout=30,
+            timeout=(DEFAULT_CONNECT_TIMEOUT_SECONDS, read_timeout),
         )
     except RequestException as exc:
         raise RemoteInvocationError(
@@ -335,6 +342,10 @@ def invoke_action(
                 )
 
             event_type = event["type"]
+            if event_type == CliEventType.PROGRESS.value and console is not None:
+                console.print_progress(event["message"])
+                continue
+
             if event_type == CliEventType.WARNING.value and console is not None:
                 console.print_warning(event["message"])
                 continue
diff --git a/tests/cli_lambda/test_app.py b/tests/cli_lambda/test_app.py
index 921a84a..8b2b2e6 100644
--- a/tests/cli_lambda/test_app.py
+++ b/tests/cli_lambda/test_app.py
@@ -32,7 +32,7 @@ def test_dispatch_action_routes_supported_actions(action: CliAction) -> None:
     handler = MagicMock(return_value=[{"type": "success"}])
 
     with patch.dict(ACTION_HANDLERS, {action: handler}, clear=True):
-        result = dispatch_action(envelope)
+        result = list(dispatch_action(envelope))
 
     assert result == [{"type": "success"}]
     handler.assert_called_once_with(envelope)
@@ -53,7 +53,7 @@ def test_execute_action_maps_handler_failures_to_terminal_error(
     handler = MagicMock(side_effect=RuntimeError("boom"))
 
     with patch.dict(ACTION_HANDLERS, {action: handler}, clear=True):
-        events = execute_action(envelope)
+        events = list(execute_action(envelope))
 
     assert events == [
         {
@@ -63,3 +63,19 @@ def test_execute_action_maps_handler_failures_to_terminal_error(
             "data": {},
         }
     ]
+
+
+def test_execute_action_maps_iteration_failures_to_terminal_error() -> None:
+    envelope = make_envelope(action=CliAction.STATUS)
+
+    def handler(_envelope):
+        yield {"type": "progress", "action": "status", "message": "started", "data": {}}
+        raise RuntimeError("stream failed")
+
+    with patch.dict(ACTION_HANDLERS, {CliAction.STATUS: handler}, clear=True):
+        events = list(execute_action(envelope))
+
+    assert events == [
+        {"type": "progress", "action": "status", "message": "started", "data": {}},
+        {"type": "error", "action": "status", "message": "stream failed", "data": {}},
+    ]
diff --git a/tests/commands/test_launch_command.py b/tests/commands/test_launch_command.py
new file mode 100644
index 0000000..c505de9
--- /dev/null
+++ b/tests/commands/test_launch_command.py
@@ -0,0 +1,159 @@
+"""Unit tests for the Lambda-backed launch command."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from devbox.cli_lambda.contracts import CliRequestEnvelope
+from devbox.cli_protocol import CliAction
+from devbox.commands.launch import (
+    MAX_USERDATA_BYTES,
+    build_launch_payload,
+    configured_param_prefix,
+    handle_launch_action,
+    read_userdata_file,
+    run_launch_command,
+    validate_launch_payload,
+)
+from devbox.launch import LaunchResult, LaunchUpdate
+
+
+def make_envelope(payload: dict | None = None) -> CliRequestEnvelope:
+    return CliRequestEnvelope(
+        version="v1",
+        action=CliAction.LAUNCH,
+        request_id="req-123",
+        param_prefix="/devbox",
+        payload=payload
+        or {
+            "project": "demo",
+            "instance_type": "t3.medium",
+            "key_pair": "demo-key",
+            "volume_size": 0,
+            "base_ami": None,
+            "userdata": None,
+            "assign_dns": True,
+            "dns_subdomain": None,
+        },
+    )
+
+
+def test_read_userdata_file_inlines_content(tmp_path) -> None:
+    path = tmp_path / "userdata.sh"
+    path.write_text("#!/bin/bash\necho hello", encoding="utf-8")
+    assert read_userdata_file(str(path)) == "#!/bin/bash\necho hello"
+
+
+def test_read_userdata_file_rejects_oversized_content(tmp_path) -> None:
+    path = tmp_path / "userdata.sh"
+    path.write_text("x" * (MAX_USERDATA_BYTES + 1), encoding="utf-8")
+    with pytest.raises(ValueError, match="must not exceed"):
+        read_userdata_file(str(path))
+
+
+def test_build_launch_payload_contains_all_options() -> None:
+    payload = build_launch_payload(
+        project="demo",
+        instance_type="t3.medium",
+        key_pair="key",
+        volume_size=100,
+        base_ami="ami-123",
+        userdata="data",
+        assign_dns=False,
+        dns_subdomain="custom",
+    )
+    assert payload == {
+        "project": "demo",
+        "instance_type": "t3.medium",
+        "key_pair": "key",
+        "volume_size": 100,
+        "base_ami": "ami-123",
+        "userdata": "data",
+        "assign_dns": False,
+        "dns_subdomain": "custom",
+    }
+
+
+@patch("devbox.commands.launch.invoke_action")
+def test_run_launch_command_uses_long_timeout_and_renders(mock_invoke, tmp_path) -> None:
+    path = tmp_path / "userdata.sh"
+    path.write_text("data", encoding="utf-8")
+    mock_invoke.return_value = {"instance_id": "i-123"}
+    console = MagicMock()
+
+    run_launch_command(
+        project="demo",
+        instance_type=None,
+        key_pair=None,
+        volume_size=0,
+        base_ami=None,
+        param_prefix="/devbox",
+        userdata_file=str(path),
+        assign_dns=True,
+        dns_subdomain=None,
+        console=console,
+    )
+
+    assert mock_invoke.call_args.kwargs["action"] is CliAction.LAUNCH
+    assert mock_invoke.call_args.kwargs["payload"]["userdata"] == "data"
+    assert mock_invoke.call_args.kwargs["read_timeout"] == 930
+    console.print_launch_result.assert_called_once_with({"instance_id": "i-123"})
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    [
+        ("project", "bad project", "alphanumeric"),
+        ("volume_size", -1, "non-negative integer"),
+        ("volume_size", True, "non-negative integer"),
+        ("assign_dns", "yes", "boolean"),
+        ("instance_type", "", "string or null"),
+        ("userdata", 123, "string or null"),
+    ],
+)
+def test_validate_launch_payload_rejects_invalid_fields(field, value, message) -> None:
+    payload = make_envelope().payload.copy()
+    payload[field] = value
+    with pytest.raises(ValueError, match=message):
+        validate_launch_payload(payload, "/devbox")
+
+
+def test_configured_param_prefix_rejects_mismatch(monkeypatch) -> None:
+    monkeypatch.setenv("DEVBOX_PARAM_PREFIX", "/configured")
+    with pytest.raises(ValueError, match="does not match"):
+        configured_param_prefix("/requested")
+
+
+@patch("devbox.commands.launch.iter_launch_workflow")
+def test_handle_launch_action_maps_workflow_events(mock_workflow, monkeypatch) -> None:
+    monkeypatch.setenv("DEVBOX_PARAM_PREFIX", "/devbox")
+    mock_workflow.return_value = iter(
+        [
+            LaunchUpdate("progress", "working"),
+            LaunchUpdate("warning", "heads up"),
+            LaunchResult(
+                project="demo",
+                instance_id="i-123",
+                state="running",
+                instance_type="t3.medium",
+                image_id="ami-123",
+                availability_zone="us-east-1a",
+                private_ip="10.0.0.1",
+                public_ip="54.0.0.1",
+                ssh_username="ubuntu",
+                dns_name=None,
+            ),
+        ]
+    )
+
+    events = list(handle_launch_action(make_envelope()))
+
+    assert [event["type"] for event in events] == [
+        "progress",
+        "warning",
+        "result",
+        "success",
+    ]
+    assert events[2]["data"]["instance_id"] == "i-123"
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 7b0e309..1d5c3bb 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -254,7 +254,7 @@ def test_launch_help(self):
         assert "--no-assign-dns" in result.output
         assert "--dns-subdomain" in result.output
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_success(self, mock_console_class, mock_launch):
         mock_console = MagicMock()
@@ -283,9 +283,10 @@ def test_launch_success(self, mock_console_class, mock_launch):
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_all_options(self, mock_console_class, mock_launch):
         mock_console = MagicMock()
@@ -320,9 +321,10 @@ def test_launch_with_all_options(self, mock_console_class, mock_launch):
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_exception(self, mock_console_class, mock_launch):
         """Test launch command with exception."""
@@ -354,7 +356,7 @@ def test_launch_missing_required_options(self):
         assert result.exit_code in [2, 3]  # Allow both Click and our error codes
         assert "Missing argument" in result.output or "Error" in result.output
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_optional_parameters_only(
         self, mock_console_class, mock_launch
@@ -377,9 +379,10 @@ def test_launch_with_optional_parameters_only(
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_instance_type_only(self, mock_console_class, mock_launch):
         """Test launch command with only instance-type specified."""
@@ -401,9 +404,10 @@ def test_launch_with_instance_type_only(self, mock_console_class, mock_launch):
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_key_pair_only(self, mock_console_class, mock_launch):
         """Test launch command with only key-pair specified."""
@@ -425,9 +429,10 @@ def test_launch_with_key_pair_only(self, mock_console_class, mock_launch):
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_dns_flags(self, mock_console_class, mock_launch):
         mock_console = MagicMock()
@@ -455,6 +460,7 @@ def test_launch_with_dns_flags(self, mock_console_class, mock_launch):
             userdata_file=None,
             assign_dns=False,
             dns_subdomain="my-custom-label",
+            console=mock_console,
         )
 
     @pytest.mark.parametrize(
@@ -466,7 +472,7 @@ def test_launch_with_dns_flags(self, mock_console_class, mock_launch):
             ("0", 0),
         ],
     )
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_volume_size_parsing(
         self, mock_console_class, mock_launch, volume_size, expected
@@ -493,7 +499,7 @@ def test_launch_volume_size_parsing(
         call_args = mock_launch.call_args
         assert call_args[1]["volume_size"] == expected
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_with_userdata_file(self, mock_console_class, mock_launch, tmp_path):
         """Test launch command with userdata file option."""
@@ -975,7 +981,7 @@ def test_status_with_realistic_data(self, mock_console_class, mock_run_status_co
             console=mock_console,
         )
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_realistic_scenario(self, mock_console_class, mock_launch):
         """Test launch command with realistic parameters."""
@@ -1011,6 +1017,7 @@ def test_launch_realistic_scenario(self, mock_console_class, mock_launch):
             userdata_file=None,  # default
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
 
@@ -1065,7 +1072,7 @@ def test_consistent_error_exit_codes(
         assert result.exit_code == 1
         mock_console.print_error.assert_called()
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_error_exit_code(self, mock_console_class, mock_launch):
         """Test launch command error exit code."""
@@ -1217,7 +1224,7 @@ def test_delete_project_uses_param_prefix_from_env(
         mock_manager_class.assert_called_once_with(prefix="env/devbox")
         mock_manager.delete_project_entry.assert_called_once_with("test-project")
 
-    @patch("devbox.launch.launch_programmatic")
+    @patch("devbox.cli.run_launch_command")
     @patch("devbox.cli.ConsoleOutput")
     def test_launch_uses_param_prefix_from_env(self, mock_console_class, mock_launch):
         mock_console = MagicMock()
@@ -1238,6 +1245,7 @@ def test_launch_uses_param_prefix_from_env(self, mock_console_class, mock_launch
             userdata_file=None,
             assign_dns=True,
             dns_subdomain=None,
+            console=mock_console,
         )
 
     @patch("devbox.new.new_project_programmatic")
diff --git a/tests/test_launch.py b/tests/test_launch.py
index c800439..37db563 100644
--- a/tests/test_launch.py
+++ b/tests/test_launch.py
@@ -225,14 +225,75 @@ def test_get_volume_info_with_min_size(mock_ec2_client):
         ]
     }
     mock_ec2_client.describe_images.return_value = mock_response
+    progress = []
 
     volumes, largest_size = get_volume_info(
-        mock_ec2_client, "ami-12345", min_volume_size=100
+        mock_ec2_client,
+        "ami-12345",
+        min_volume_size=100,
+        on_progress=progress.append,
     )
 
     assert len(volumes) == 1
     assert largest_size == 100  # Should be adjusted to minimum
     assert volumes[0]["Ebs"]["VolumeSize"] == 100
+    assert progress == ["Increasing volume size from 20 GiB to 100 GiB"]
+
+
+def test_get_volume_info_creates_volume_with_progress(mock_ec2_client):
+    """Test creation of a volume when the AMI has no EBS mapping."""
+    mock_ec2_client.describe_images.return_value = {
+        "Images": [{"BlockDeviceMappings": []}]
+    }
+    progress = []
+
+    volumes, largest_size = get_volume_info(
+        mock_ec2_client,
+        "ami-12345",
+        min_volume_size=50,
+        on_progress=progress.append,
+    )
+
+    assert volumes == [
+        {
+            "DeviceName": "/dev/sda1",
+            "Ebs": {
+                "VolumeSize": 50,
+                "VolumeType": "gp3",
+                "Encrypted": True,
+                "DeleteOnTermination": True,
+            },
+        }
+    ]
+    assert largest_size == 50
+    assert progress == ["Creating new volume of size 50 GiB"]
+
+
+def test_get_volume_info_without_adjustment_emits_no_progress(mock_ec2_client):
+    """Test an already-large volume does not produce adjustment progress."""
+    mock_ec2_client.describe_images.return_value = {
+        "Images": [
+            {
+                "BlockDeviceMappings": [
+                    {
+                        "DeviceName": "/dev/sda1",
+                        "Ebs": {"VolumeSize": 100, "VolumeType": "gp3"},
+                    }
+                ]
+            }
+        ]
+    }
+    progress = []
+
+    _, largest_size = get_volume_info(
+        mock_ec2_client,
+        "ami-12345",
+        min_volume_size=50,
+        on_progress=progress.append,
+    )
+
+    assert largest_size == 100
+    assert progress == []
 
 
 def test_get_volume_info_no_image(mock_ec2_client):
@@ -285,6 +346,87 @@ def test_get_launch_template_info_with_errors(mock_ec2_client):
     assert "name" in az_info["lt-12345"]
 
 
+def test_get_launch_template_info_warns_on_template_lookup_failure(
+    mock_ec2_client,
+):
+    """Test template lookup errors warn while retaining the generic AZ label."""
+    mock_ec2_client.describe_launch_templates.side_effect = ClientError(
+        {
+            "Error": {
+                "Code": "UnauthorizedOperation",
+                "Message": "not allowed to inspect template",
+            }
+        },
+        "DescribeLaunchTemplates",
+    )
+    warnings = []
+
+    az_info = get_launch_template_info(
+        mock_ec2_client, ["lt-12345"], on_warning=warnings.append
+    )
+
+    assert az_info["lt-12345"] == {"name": "az-1", "index": "1"}
+    assert warnings == [
+        "Unable to inspect launch template lt-12345: UnauthorizedOperation: "
+        "not allowed to inspect template; using fallback availability-zone "
+        "label az-1"
+    ]
+
+
+def test_get_launch_template_info_warns_on_subnet_lookup_failure(mock_ec2_client):
+    """Test subnet errors warn while retaining the AZ parsed from the template."""
+    mock_ec2_client.describe_launch_templates.return_value = {
+        "LaunchTemplates": [{"LaunchTemplateName": "devbox-us-east-1b-template"}]
+    }
+    mock_ec2_client.describe_launch_template_versions.return_value = {
+        "LaunchTemplateVersions": [
+            {
+                "LaunchTemplateData": {
+                    "NetworkInterfaces": [{"SubnetId": "subnet-12345"}]
+                }
+            }
+        ]
+    }
+    mock_ec2_client.describe_subnets.side_effect = ClientError(
+        {
+            "Error": {
+                "Code": "InvalidSubnetID.NotFound",
+                "Message": "subnet is not visible",
+            }
+        },
+        "DescribeSubnets",
+    )
+    warnings = []
+
+    az_info = get_launch_template_info(
+        mock_ec2_client, ["lt-12345"], on_warning=warnings.append
+    )
+
+    assert az_info["lt-12345"] == {"name": "us-east-1b", "index": "2"}
+    assert warnings == [
+        "Unable to inspect subnet subnet-12345 for launch template lt-12345: "
+        "InvalidSubnetID.NotFound: subnet is not visible; using "
+        "availability-zone label us-east-1b"
+    ]
+
+
+def test_get_launch_template_info_logs_warning_without_callback(
+    mock_ec2_client, caplog
+):
+    """Test best-effort metadata failures are never silently discarded."""
+    mock_ec2_client.describe_launch_templates.side_effect = ClientError(
+        {"Error": {"Code": "RequestLimitExceeded", "Message": "try later"}},
+        "DescribeLaunchTemplates",
+    )
+
+    get_launch_template_info(mock_ec2_client, ["lt-12345"])
+
+    assert (
+        "Unable to inspect launch template lt-12345: RequestLimitExceeded: "
+        "try later; using fallback availability-zone label az-1" in caplog.text
+    )
+
+
 # Test functions for launch_instance
 
 
diff --git a/tests/test_launch_workflow.py b/tests/test_launch_workflow.py
new file mode 100644
index 0000000..6acc19f
--- /dev/null
+++ b/tests/test_launch_workflow.py
@@ -0,0 +1,379 @@
+"""Tests for the structured launch workflow."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from botocore.exceptions import ClientError
+
+from devbox.launch import LaunchRequest, LaunchResult, LaunchUpdate, iter_launch_workflow
+
+
+def make_instance(*states: str) -> MagicMock:
+    """Build an EC2 instance resource that advances through states on reload."""
+    instance = MagicMock()
+    remaining = iter(states)
+
+    def reload() -> None:
+        state = next(remaining, states[-1])
+        instance.state = {"Name": state}
+        instance.meta.data = {
+            "InstanceId": "i-123",
+            "State": {"Name": state},
+            "InstanceType": "t3.medium",
+            "ImageId": "ami-123",
+            "Placement": {"AvailabilityZone": "us-east-1a"},
+            "PrivateIpAddress": "10.0.0.1",
+            "PublicIpAddress": "54.0.0.1",
+        }
+
+    instance.meta.data = {"State": {"Name": states[0]}}
+    instance.public_dns_name = "ec2.example.internal"
+    instance.reload.side_effect = reload
+    return instance
+
+
+def make_aws() -> dict[str, MagicMock]:
+    return {
+        "ssm": MagicMock(),
+        "ddb": MagicMock(),
+        "ec2": MagicMock(),
+        "ec2_resource": MagicMock(),
+    }
+
+
+def make_client_error(code: str) -> ClientError:
+    """Build a representative EC2 client error."""
+    return ClientError(
+        {"Error": {"Code": code, "Message": code}},
+        "DescribeInstances",
+    )
+
+
+@patch("devbox.launch.update_instance_status")
+@patch("devbox.launch._resolve_ssh_username", return_value=("ubuntu", None))
+@patch("devbox.launch.DNSManager.from_ssm")
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info")
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_emits_updates_and_result(
+    mock_config,
+    mock_volumes,
+    mock_templates,
+    mock_launch,
+    mock_dns_from_ssm,
+    _mock_username,
+    mock_update,
+) -> None:
+    aws = make_aws()
+    table = MagicMock()
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-123"],
+        "table": table,
+    }
+    def get_volumes(_ec2, _image_id, _minimum, *, on_progress):
+        on_progress("Increasing volume size from 20 GiB to 50 GiB")
+        return [], 50
+
+    def get_templates(_ec2, _lt_ids, *, on_warning):
+        on_warning(
+            "Unable to inspect launch template lt-123: AccessDenied: denied; "
+            "using fallback availability-zone label us-east-1a"
+        )
+        return {"lt-123": {"name": "us-east-1a", "index": "1"}}
+
+    mock_volumes.side_effect = get_volumes
+    mock_templates.side_effect = get_templates
+    instance = make_instance("pending", "running", "running")
+    mock_launch.return_value = (instance, "i-123", None)
+    dns_manager = MagicMock()
+    dns_manager.provider = object()
+    dns_manager.sanitize_dns_name.return_value = "demo"
+    dns_manager.assign_cname.return_value = "demo.example.com"
+    mock_dns_from_ssm.return_value = dns_manager
+    sleeper = MagicMock()
+
+    events = list(
+        iter_launch_workflow(
+            LaunchRequest(
+                project="demo",
+                instance_type="t3.medium",
+                key_pair="demo-key",
+                userdata="#!/bin/bash",
+            ),
+            aws=aws,
+            sleeper=sleeper,
+        )
+    )
+
+    assert any(
+        isinstance(event, LaunchUpdate) and "waiting for running state" in event.message
+        for event in events
+    )
+    volume_progress_index = events.index(
+        LaunchUpdate("progress", "Increasing volume size from 20 GiB to 50 GiB")
+    )
+    template_warning_index = events.index(
+        LaunchUpdate(
+            "warning",
+            "Unable to inspect launch template lt-123: AccessDenied: denied; "
+            "using fallback availability-zone label us-east-1a",
+        )
+    )
+    launch_attempt_index = events.index(
+        LaunchUpdate("progress", "Attempting to launch instance in us-east-1a")
+    )
+    assert volume_progress_index < template_warning_index < launch_attempt_index
+    result = events[-1]
+    assert result == LaunchResult(
+        project="demo",
+        instance_id="i-123",
+        state="running",
+        instance_type="t3.medium",
+        image_id="ami-123",
+        availability_zone="us-east-1a",
+        private_ip="10.0.0.1",
+        public_ip="54.0.0.1",
+        ssh_username="ubuntu",
+        dns_name="demo.example.com",
+    )
+    assert mock_launch.call_args.kwargs["userdata"] == "#!/bin/bash"
+    sleeper.assert_called_once_with(15)
+    mock_update.assert_called_once()
+
+
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_uses_saved_instance_type_and_key_pair(mock_config) -> None:
+    mock_config.return_value = {
+        "item": {
+            "Status": "READY",
+            "AMI": "ami-123",
+            "LastInstanceType": "m7i.large",
+            "LastKeyPair": "saved-key",
+        },
+        "lt_ids": [],
+        "table": MagicMock(),
+    }
+
+    with pytest.raises(RuntimeError, match="all availability zones"):
+        list(iter_launch_workflow(LaunchRequest(project="demo"), aws=make_aws()))
+
+
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info", return_value=([], 0))
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_reports_multi_az_failures(
+    mock_config,
+    _mock_volumes,
+    mock_templates,
+    mock_launch,
+) -> None:
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-a", "lt-b"],
+        "table": MagicMock(),
+    }
+    mock_templates.return_value = {
+        "lt-a": {"name": "us-east-1a", "index": "1"},
+        "lt-b": {"name": "us-east-1b", "index": "2"},
+    }
+    instance = make_instance("running", "running")
+    mock_launch.side_effect = [
+        (None, None, RuntimeError("capacity")),
+        (instance, "i-123", None),
+    ]
+
+    with (
+        patch("devbox.launch.DNSManager.from_ssm") as mock_dns,
+        patch("devbox.launch._resolve_ssh_username", return_value=(None, "unknown")),
+        patch("devbox.launch.update_instance_status"),
+    ):
+        mock_dns.return_value.provider = None
+        events = list(
+            iter_launch_workflow(
+                LaunchRequest(
+                    project="demo",
+                    instance_type="t3.medium",
+                    key_pair="key",
+                ),
+                aws=make_aws(),
+            )
+        )
+
+    assert any(
+        isinstance(event, LaunchUpdate)
+        and event.kind == "warning"
+        and "capacity" in event.message
+        for event in events
+    )
+
+
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info", return_value=([], 0))
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_times_out_waiting_for_running(
+    mock_config,
+    _mock_volumes,
+    mock_templates,
+    mock_launch,
+) -> None:
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-a"],
+        "table": MagicMock(),
+    }
+    mock_templates.return_value = {"lt-a": {"name": "us-east-1a", "index": "1"}}
+    mock_launch.return_value = (make_instance("pending"), "i-123", None)
+
+    with pytest.raises(TimeoutError, match="did not reach running state"):
+        list(
+            iter_launch_workflow(
+                LaunchRequest(
+                    project="demo", instance_type="t3.medium", key_pair="key"
+                ),
+                aws=make_aws(),
+                sleeper=lambda _seconds: None,
+                max_poll_attempts=2,
+            )
+        )
+
+
+@patch("devbox.launch.update_instance_status")
+@patch("devbox.launch._resolve_ssh_username", return_value=("ubuntu", None))
+@patch("devbox.launch.DNSManager.from_ssm")
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info", return_value=([], 0))
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_retries_instance_not_found(
+    mock_config,
+    _mock_volumes,
+    mock_templates,
+    mock_launch,
+    mock_dns,
+    _mock_username,
+    _mock_update,
+) -> None:
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-a"],
+        "table": MagicMock(),
+    }
+    mock_templates.return_value = {"lt-a": {"name": "us-east-1a", "index": "1"}}
+    instance = make_instance("running")
+    normal_reload = instance.reload.side_effect
+    calls = 0
+
+    def reload() -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise make_client_error("InvalidInstanceID.NotFound")
+        normal_reload()
+
+    instance.reload.side_effect = reload
+    mock_launch.return_value = (instance, "i-123", None)
+    mock_dns.return_value.provider = None
+    sleeper = MagicMock()
+
+    events = list(
+        iter_launch_workflow(
+            LaunchRequest(project="demo", instance_type="t3.medium", key_pair="key"),
+            aws=make_aws(),
+            sleeper=sleeper,
+        )
+    )
+
+    assert isinstance(events[-1], LaunchResult)
+    assert any(
+        isinstance(event, LaunchUpdate)
+        and "not yet visible to EC2" in event.message
+        for event in events
+    )
+    sleeper.assert_called_once_with(15)
+
+
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info", return_value=([], 0))
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_times_out_when_instance_remains_not_found(
+    mock_config,
+    _mock_volumes,
+    mock_templates,
+    mock_launch,
+) -> None:
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-a"],
+        "table": MagicMock(),
+    }
+    mock_templates.return_value = {"lt-a": {"name": "us-east-1a", "index": "1"}}
+    instance = MagicMock()
+    instance.meta.data = {"State": {"Name": "pending"}}
+    instance.reload.side_effect = make_client_error("InvalidInstanceID.NotFound")
+    mock_launch.return_value = (instance, "i-123", None)
+
+    with pytest.raises(TimeoutError, match="did not reach running state"):
+        list(
+            iter_launch_workflow(
+                LaunchRequest(
+                    project="demo", instance_type="t3.medium", key_pair="key"
+                ),
+                aws=make_aws(),
+                sleeper=lambda _seconds: None,
+                max_poll_attempts=2,
+            )
+        )
+
+
+@patch("devbox.launch.launch_instance")
+@patch("devbox.launch.get_launch_template_info")
+@patch("devbox.launch.get_volume_info", return_value=([], 0))
+@patch("devbox.launch.get_launch_config")
+def test_iter_launch_workflow_propagates_other_reload_errors(
+    mock_config,
+    _mock_volumes,
+    mock_templates,
+    mock_launch,
+) -> None:
+    mock_config.return_value = {
+        "item": {"Status": "READY", "AMI": "ami-123"},
+        "lt_ids": ["lt-a"],
+        "table": MagicMock(),
+    }
+    mock_templates.return_value = {"lt-a": {"name": "us-east-1a", "index": "1"}}
+    instance = MagicMock()
+    instance.meta.data = {"State": {"Name": "pending"}}
+    instance.reload.side_effect = make_client_error("UnauthorizedOperation")
+    mock_launch.return_value = (instance, "i-123", None)
+
+    with pytest.raises(ClientError) as exc_info:
+        list(
+            iter_launch_workflow(
+                LaunchRequest(
+                    project="demo", instance_type="t3.medium", key_pair="key"
+                ),
+                aws=make_aws(),
+                sleeper=lambda _seconds: None,
+            )
+        )
+
+    assert exc_info.value.response["Error"]["Code"] == "UnauthorizedOperation"
+
+
+@pytest.mark.parametrize(
+    "launch_request,message",
+    [
+        (LaunchRequest(project="bad project"), "Project name"),
+        (LaunchRequest(project="demo", volume_size=-1), "Volume size"),
+    ],
+)
+def test_iter_launch_workflow_validates_inputs(launch_request, message) -> None:
+    with pytest.raises(ValueError, match=message):
+        list(iter_launch_workflow(launch_request, aws=make_aws()))
diff --git a/tests/test_remote_client.py b/tests/test_remote_client.py
index f2ca9a6..55544bf 100644
--- a/tests/test_remote_client.py
+++ b/tests/test_remote_client.py
@@ -130,6 +130,43 @@ def test_invoke_action_surfaces_warning_events():
     console.print_warning.assert_called_once_with("heads-up")
 
 
+@responses.activate
+def test_invoke_action_surfaces_progress_events():
+    console = MagicMock()
+    responses.add(
+        responses.POST,
+        FUNCTION_URL,
+        body=(
+            '{"type":"progress","action":"demo-action","message":"working","data":{}}\n'
+            '{"type":"result","action":"demo-action","message":"ready","data":{"ok":true}}\n'
+            '{"type":"success","action":"demo-action","message":"done","data":{}}\n'
+        ),
+        content_type=NDJSON_MIME_TYPE,
+        status=200,
+    )
+
+    with patch("devbox.remote_client.utils.get_ssm_parameter", return_value=FUNCTION_URL):
+        invoke_action("demo-action", {}, "/devbox", console=console)
+
+    console.print_progress.assert_called_once_with("working")
+
+
+def test_invoke_action_uses_configurable_timeout():
+    response = MagicMock(status_code=200)
+    response.iter_lines.return_value = [
+        '{"type":"result","action":"demo-action","message":"ready","data":{"ok":true}}',
+        '{"type":"success","action":"demo-action","message":"done","data":{}}',
+    ]
+    with (
+        patch("devbox.remote_client.utils.get_ssm_parameter", return_value=FUNCTION_URL),
+        patch("devbox.remote_client.requests.post", return_value=response) as mock_post,
+    ):
+        invoke_action("demo-action", {}, "/devbox", read_timeout=930)
+
+    assert mock_post.call_args.kwargs["timeout"] == (10, 930)
+    response.close.assert_called_once_with()
+
+
 @responses.activate
 def test_invoke_action_rejects_malformed_ndjson():
     responses.add(