diff --git a/features/plan_cli_coverage_boost.feature b/features/plan_cli_coverage_boost.feature
index 93e9332db..ffeaf60ee 100644
--- a/features/plan_cli_coverage_boost.feature
+++ b/features/plan_cli_coverage_boost.feature
@@ -64,6 +64,14 @@ Feature: Plan CLI coverage boost
     And the plan coverage output should contain "command"
     And the plan coverage output should contain "exit_code"
 
+  Scenario: execute_plan rich output shows spec panels
+    Given a plan lifecycle CLI runner for coverage
+    And a mocked lifecycle service for plan coverage commands
+    And the service has a spec-compliant execute plan for rich output
+    When I invoke execute in rich format with plan id
+    Then the plan coverage command should succeed
+    And the execute rich output shows spec panels
+
 #   @tdd_issue @tdd_issue_4251 @tdd_expected_fail @skip
   @skip
   Scenario: execute_plan JSON output has spec-required envelope structure
diff --git a/features/plan_lifecycle_cli_coverage.feature b/features/plan_lifecycle_cli_coverage.feature
index 39ce20165..6173d8bad 100644
--- a/features/plan_lifecycle_cli_coverage.feature
+++ b/features/plan_lifecycle_cli_coverage.feature
@@ -47,7 +47,7 @@ Feature: Plan lifecycle CLI coverage
     When I run plan execute without a plan id with 1 complete plans
     Then the plan lifecycle command should succeed
     And the execute command should run the single ready plan
-    And the plan lifecycle output should contain "Plan Executed"
+    And the plan lifecycle output should contain "Execution started"
 
   Scenario: Plan execute handles invalid phase transition
     When I run plan execute for plan id "01ARZ3NDEKTSV4RRFFQ69G5FAV" causing "invalid transition"
diff --git a/features/steps/plan_cli_coverage_boost_steps.py b/features/steps/plan_cli_coverage_boost_steps.py
index 2318c9584..511b70968 100644
--- a/features/steps/plan_cli_coverage_boost_steps.py
+++ b/features/steps/plan_cli_coverage_boost_steps.py
@@ -29,9 +29,11 @@ from cleveragents.cli.commands.plan import (
     app as plan_app,
 )
 from cleveragents.domain.models.core.plan import (
+    InvariantSource,
     NamespacedName,
     Plan,
     PlanIdentity,
+    PlanInvariant,
     PlanPhase,
     PlanTimestamps,
     ProcessingState,
@@ -74,6 +76,10 @@ def _make_plan(
         namespaced_name=NamespacedName.parse(name),
         action_name=action_name,
         description=description,
+        definition_of_done=None,
+        strategy_actor=None,
+        execution_actor=None,
+        created_by=None,
         phase=phase,
         processing_state=processing_state,
         project_links=project_links or [],
@@ -306,6 +312,62 @@ def step_service_has_strategize_plan(context) -> None:
     context._cleanup_handlers.append(executor_patcher.stop)
 
 
+@given("the service has a spec-compliant execute plan for rich output")
+def step_service_has_rich_execute_plan(context) -> None:
+    plan_id = _ULIDS[0]
+    strategize_complete = _make_plan(
+        plan_id=plan_id,
+        name="local/exec-rich-plan",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.COMPLETE,
+    )
+    execute_queued = _make_plan(
+        plan_id=plan_id,
+        name="local/exec-rich-plan",
+        phase=PlanPhase.EXECUTE,
+        processing_state=ProcessingState.QUEUED,
+    )
+    execute_complete = _make_plan(
+        plan_id=plan_id,
+        name="local/exec-rich-plan",
+        phase=PlanPhase.EXECUTE,
+        processing_state=ProcessingState.COMPLETE,
+    )
+
+    sandbox_path = f"/repos/api/.worktrees/plan-{plan_id[:8].lower()}"
+    for idx, plan in enumerate((execute_queued, execute_complete), start=1):
+        plan.sandbox_refs = [sandbox_path]
+        plan.strategy_actor = "local/senior-planner"
+        plan.execution_actor = "local/executor"
+        plan.decisions = [
+            {"id": f"d{idx}a"},
+            {"id": f"d{idx}b"},
+            {"id": f"d{idx}c"},
+        ]
+        plan.invariants = [
+            PlanInvariant(text="Keep tests passing", source=InvariantSource.ACTION),
+            PlanInvariant(text="Document work", source=InvariantSource.PROJECT),
+        ]
+        plan.timestamps.execute_started_at = datetime(2026, 1, 1, 12, 58, 10)
+
+    execute_complete.timestamps.execute_completed_at = datetime(2026, 1, 1, 13, 0, 0)
+
+    context.mock_lifecycle_service.get_plan.side_effect = [
+        strategize_complete,
+        execute_queued,
+        execute_complete,
+    ]
+    context.mock_lifecycle_service.execute_plan.return_value = execute_queued
+    context._execute_plan_id = plan_id
+
+    executor_patcher = patch(
+        "cleveragents.cli.commands.plan._get_plan_executor",
+        return_value=MagicMock(),
+    )
+    executor_patcher.start()
+    context._cleanup_handlers.append(executor_patcher.stop)
+
+
 @given("the service has a complete execute plan for apply")
 def step_service_has_execute_plan(context) -> None:
     pre_plan = _make_plan(
@@ -408,6 +470,14 @@ def step_invoke_execute_json(context) -> None:
     )
 
 
+@when("I invoke execute in rich format with plan id")
+def step_invoke_execute_rich(context) -> None:
+    context.result = context.runner.invoke(
+        plan_app,
+        ["execute", context._execute_plan_id],
+    )
+
+
 @when('I invoke apply with "--format" "json" and plan id')
 def step_invoke_apply_json(context) -> None:
     context.result = context.runner.invoke(
@@ -519,6 +589,24 @@ def step_plan_coverage_output_contains(context, text: str) -> None:
     assert text in output, f"Expected '{text}' in output:\n{output}"
 
 
+@then("the execute rich output shows spec panels")
+def step_execute_rich_output_panels(context) -> None:
+    output = _output(context)
+    expected = [
+        "Execution",
+        "Sandbox",
+        "Strategy Summary",
+        "Progress",
+        "Collect context",
+        "Run tools",
+        "Build changeset",
+        "Validate",
+        "Execution started",
+    ]
+    for text in expected:
+        assert text in output, f"Expected '{text}' in rich output:\n{output}"
+
+
 @then('the plan coverage output should not contain "{text}"')
 def step_plan_coverage_output_not_contains(context, text: str) -> None:
     output = _output(context)
diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py
index f3a249d6b..694fa35a3 100644
--- a/src/cleveragents/cli/commands/plan.py
+++ b/src/cleveragents/cli/commands/plan.py
@@ -28,7 +28,7 @@ import warnings
 from contextlib import suppress
 from datetime import datetime
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any, cast
 
 import typer
 from rich.console import Console
@@ -471,6 +471,156 @@ def _execute_output_dict(
     }
 
 
+def _print_execute_rich_output(plan: Any, envelope: dict[str, object]) -> None:
+    """Render rich output panels for ``plan execute`` according to the spec."""
+
+    data = envelope.get("data", {}) if isinstance(envelope, dict) else {}
+    if not isinstance(data, dict):
+        data = {}
+
+    sandbox_dict = data.get("sandbox") if isinstance(data, dict) else {}
+    if not isinstance(sandbox_dict, dict):
+        sandbox_dict = {}
+
+    strategy_summary = data.get("strategy_summary") if isinstance(data, dict) else {}
+    if not isinstance(strategy_summary, dict):
+        strategy_summary = {}
+
+    progress_entries = data.get("progress") if isinstance(data, dict) else []
+    progress_list: list[dict[str, str]] = []
+    if isinstance(progress_entries, list):
+        for entry in progress_entries:
+            if isinstance(entry, dict):
+                progress_list.append(entry)
+
+    plan_id = data.get("plan_id") if isinstance(data, dict) else None
+    if not isinstance(plan_id, str) or not plan_id:
+        plan_id = getattr(getattr(plan, "identity", None), "plan_id", "-")
+
+    phase_value = data.get("phase") if isinstance(data, dict) else None
+    if not isinstance(phase_value, str) or not phase_value:
+        phase_value = getattr(getattr(plan, "phase", None), "value", "-")
+
+    worker_value = data.get("worker") if isinstance(data, dict) else None
+    if not isinstance(worker_value, str) or not worker_value:
+        worker_value = getattr(plan, "execution_actor", None) or "local/executor"
+
+    started_value = data.get("started") if isinstance(data, dict) else None
+    attempt_value = data.get("attempt") if isinstance(data, dict) else None
+
+    execution_rows: list[tuple[str, str]] = [
+        ("[cyan bold]Plan:[/cyan bold]", plan_id),
+        ("[yellow bold]Phase:[/yellow bold]", phase_value),
+        (
+            "[magenta bold]Sandbox:[/magenta bold]",
+            sandbox_dict.get("strategy") or "git_worktree",
+        ),
+        ("[#5599ff bold]Worker:[/#5599ff bold]", worker_value),
+    ]
+    if isinstance(started_value, str) and started_value:
+        execution_rows.append(("[green bold]Started:[/green bold]", started_value))
+    if isinstance(attempt_value, int):
+        execution_rows.append(
+            ("[#5599ff bold]Attempt:[/#5599ff bold]", str(attempt_value))
+        )
+
+    _print_panel("Execution", execution_rows)
+
+    sandbox_rows: list[tuple[str, str]] = [
+        (
+            "[#5599ff bold]Strategy:[/#5599ff bold]",
+            sandbox_dict.get("strategy") or "git_worktree",
+        ),
+        (
+            "[#5599ff bold]Path:[/#5599ff bold]",
+            sandbox_dict.get("path") or "—",
+        ),
+        (
+            "[#5599ff bold]Branch:[/#5599ff bold]",
+            sandbox_dict.get("branch") or "—",
+        ),
+        (
+            "[green bold]Status:[/green bold]",
+            sandbox_dict.get("status") or "unknown",
+        ),
+    ]
+    _print_panel("Sandbox", sandbox_rows)
+
+    strategy_rows: list[tuple[str, str]] = [
+        (
+            "[#5599ff bold]Decisions:[/#5599ff bold]",
+            str(strategy_summary.get("decisions", 0)),
+        ),
+        (
+            "[magenta bold]Invariants:[/magenta bold]",
+            str(strategy_summary.get("invariants", 0)),
+        ),
+        (
+            "[#5599ff bold]Planned Child Plans:[/#5599ff bold]",
+            str(strategy_summary.get("planned_child_plans", 0)),
+        ),
+        (
+            "[#5599ff bold]Estimated Files:[/#5599ff bold]",
+            str(strategy_summary.get("estimated_files", 0)),
+        ),
+        (
+            "[#5599ff bold]Risk:[/#5599ff bold]",
+            str(strategy_summary.get("risk", "unknown")),
+        ),
+    ]
+    _print_panel("Strategy Summary", strategy_rows)
+
+    progress_lines: list[str] = []
+    if progress_list:
+        for entry in progress_list:
+            label = entry.get("label")
+            if not isinstance(label, str):
+                label = str(label)
+            status = entry.get("status")
+            status_str = status if isinstance(status, str) else str(status)
+            symbol, color = _progress_symbol(status_str)
+            progress_lines.append(f"[{color}]{symbol}[/{color}] {label}")
+    else:
+        progress_lines.append("[dim]No progress reported[/dim]")
+
+    progress_table = Table.grid(padding=0)
+    for line in progress_lines:
+        progress_table.add_row(line)
+    console.print(Panel(progress_table, title="Progress", expand=False))
+
+    messages = envelope.get("messages") if isinstance(envelope, dict) else None
+    message_text = "Execution started"
+    if isinstance(messages, list) and messages:
+        first_message = messages[0]
+        if isinstance(first_message, str) and first_message:
+            message_text = first_message
+    console.print(f"[green]✓ OK[/green] {message_text}")
+
+
+def _print_panel(title: str, rows: list[tuple[str, str]]) -> None:
+    """Render a two-column Rich panel from rows."""
+
+    table = Table.grid(padding=(0, 1))
+    table.add_column(justify="left")
+    table.add_column(justify="left")
+    for label, value in rows:
+        table.add_row(label, str(value))
+    console.print(Panel(table, title=title, expand=False))
+
+
+def _progress_symbol(status: str) -> tuple[str, str]:
+    """Map a progress status to a symbol and colour."""
+
+    normalised = status.lower()
+    if normalised == "complete":
+        return "✓", "green"
+    if normalised == "running":
+        return "⏳", "cyan"
+    if normalised == "error":
+        return "✗", "red"
+    return "•", "yellow"
+
+
 # Programmatic wrapper functions for testing and scripting
 def tell_command(prompt: str, name: str | None = None) -> None:
     """Programmatic interface for creating a plan from instructions.
@@ -2451,32 +2601,18 @@ def execute_plan(
                     plan_id,
                 )
 
-        if fmt != OutputFormat.RICH.value:
-            execute_elapsed_ms = int(
-                (datetime.now() - execute_wall_start).total_seconds() * 1000
-            )
-            envelope = _execute_output_dict(
-                plan,
-                started_at=execute_wall_start,
-                duration_ms=execute_elapsed_ms,
-            )
-            console.print(format_output(envelope, fmt))
+        execute_elapsed_ms = int(
+            (datetime.now() - execute_wall_start).total_seconds() * 1000
+        )
+        envelope = _execute_output_dict(
+            plan,
+            started_at=execute_wall_start,
+            duration_ms=execute_elapsed_ms,
+        )
+        if fmt == OutputFormat.RICH.value:
+            _print_execute_rich_output(plan, envelope)
         else:
-            _print_lifecycle_plan(plan, title="Plan Executed")
-            phase_label = f"{plan.phase.value}/{plan.state.value}"
-            if plan.phase == PlanPhase.EXECUTE and plan.state in (
-                ProcessingState.COMPLETE,
-                ProcessingState.APPLIED,
-            ):
-                console.print(
-                    f"\n[dim]Plan execution completed ({phase_label}). "
-                    "Run 'agents plan apply <id>' when ready.[/dim]"
-                )
-            else:
-                console.print(
-                    f"\n[dim]Plan is now in {phase_label} state. "
-                    "Run 'agents plan execute <id>' to continue.[/dim]"
-                )
+            console.print(format_output(envelope, fmt))
 
     except PreflightRejection as e:
         console.print(f"[red]Pre-flight check failed:[/red] {e}")
@@ -4320,7 +4456,8 @@ def build_decision_tree(
     for rid in roots:
         node = _node_dict(by_id[rid])
         result.append(node)
-        queue.append((rid, node["children"], 1))  # type: ignore[arg-type]  # children value is list at runtime; dict[str, object] prevents narrowing
+        children_list = cast(list[dict[str, object]], node["children"])
+        queue.append((rid, children_list, 1))
 
     while queue:
         did, parent_list, depth_val = queue.popleft()
@@ -4331,9 +4468,8 @@ def build_decision_tree(
                 continue
             child_node = _node_dict(by_id[child_id])
             parent_list.append(child_node)
-            queue.append(
-                (child_id, child_node["children"], depth_val + 1)  # type: ignore[arg-type]  # children value is list at runtime; dict[str, object] prevents narrowing
-            )
+            child_children = cast(list[dict[str, object]], child_node["children"])
+            queue.append((child_id, child_children, depth_val + 1))
 
     return result