CHANGELOG.md

@@ -129,6 +129,13 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   `pr-merge-pool-supervisor` to the product-builder's supervisor launch list (18 total
   supervisors). Updated all numeric references, pre-flight checklists, and validation logic.
 
+### Fixed
+
+- **Plan Execute Rich Output** (#6344): Restored the spec-required Execution, Sandbox,
+  Strategy Summary, and Progress panels in the CLI rich output, ensured the Started and
+  Attempt rows always render with safe fallbacks, and added Robot integration coverage to
+  guard the behaviour going forward.
+
 ---
 
 ## [3.8.0] — 2026-04-05

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

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"

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)

robot/cli_lifecycle_e2e.robot

@@ -32,6 +32,14 @@ Plan Execute Transitions To Execute Phase
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    cli-lifecycle-plan-execute-ok
 
+Plan Execute Rich Output Shows Panels End-To-End
+    [Documentation]    Verify rich plan execute output renders spec-required panels with fallback values
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    plan-execute-rich-panels    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-lifecycle-plan-execute-rich-panels-ok
+
 Plan Lifecycle Apply Transitions To Apply Phase
     [Documentation]    Apply a plan, transitioning from Execute to Apply
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    plan-apply    cwd=${WORKSPACE}

robot/helper_cli_lifecycle_e2e.py

@@ -6,8 +6,10 @@ Each subcommand is self-contained and prints a sentinel on success.
 from __future__ import annotations
 
 import os
+import re
 import sys
 import tempfile
+import types
 from datetime import datetime
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -17,6 +19,207 @@ _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
     sys.path.insert(0, _SRC)
 
+# Provide a lightweight structlog stub so command modules import without optional dep
+if "structlog" not in sys.modules:
+    sys.modules["structlog"] = types.SimpleNamespace(
+        get_logger=lambda *args, **kwargs: MagicMock(name="structlog_logger")
+    )
+
+if "dependency_injector" not in sys.modules:
+    di_module = types.ModuleType("dependency_injector")
+    containers_module = types.ModuleType("dependency_injector.containers")
+    providers_module = types.ModuleType("dependency_injector.providers")
+
+    class _Provider:
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+            self._value = MagicMock(name="dependency_injector_provider")
+
+        def override(self, other):
+            if isinstance(other, _Provider):
+                self._value = other._value
+            else:
+                self._value = other
+            return self
+
+        def __call__(self, *args, **kwargs):
+            if callable(self._value):
+                return self._value(*args, **kwargs)
+            return self._value
+
+    class DeclarativeContainer:
+        pass
+
+    def _make_provider(value=None):
+        provider = _Provider()
+        if value is not None:
+            provider._value = value
+        return provider
+
+    def Singleton(*args, **kwargs):
+        return _Provider(*args, **kwargs)
+
+    def Factory(*args, **kwargs):
+        return _Provider(*args, **kwargs)
+
+    def Callable(func, *args, **kwargs):
+        provider = _Provider(*args, **kwargs)
+        provider._value = func
+        return provider
+
+    def Object(value):
+        return _make_provider(value)
+
+    def Configuration(*args, **kwargs):
+        return _Provider(*args, **kwargs)
+
+    containers_module.DeclarativeContainer = DeclarativeContainer
+    providers_module.Provider = _Provider
+    providers_module.Singleton = Singleton
+    providers_module.Factory = Factory
+    providers_module.Callable = Callable
+    providers_module.Object = Object
+    providers_module.Configuration = Configuration
+
+    di_module.containers = containers_module
+    di_module.providers = providers_module
+
+    sys.modules["dependency_injector"] = di_module
+    sys.modules["dependency_injector.containers"] = containers_module
+    sys.modules["dependency_injector.providers"] = providers_module
+
+if "ulid" not in sys.modules:
+
+    class _ULID:
+        def __init__(self, value: str | None = None):
+            self._value = value or "01HJ0000000000000000000000"
+
+        def __str__(self) -> str:
+            return self._value
+
+        def __repr__(self) -> str:
+            return f"ULID('{self._value}')"
+
+    sys.modules["ulid"] = types.SimpleNamespace(ULID=_ULID)
+
+if "cleveragents.application.container" not in sys.modules:
+    container_stub = types.ModuleType("cleveragents.application.container")
+
+    def _stub_get_container(*args, **kwargs):  # pragma: no cover - defensive stub
+        raise RuntimeError("container stub should not be invoked in helper tests")
+
+    container_stub.get_container = _stub_get_container
+    sys.modules["cleveragents.application.container"] = container_stub
+
+if "rx" not in sys.modules:
+    rx_module = types.ModuleType("rx")
+    rx_module.__path__ = []
+
+    core_module = types.ModuleType("rx.core")
+    core_module.__path__ = []
+    core_observable_module = types.ModuleType("rx.core.observable")
+    core_observable_module.__path__ = []
+    core_observable_submodule = types.ModuleType("rx.core.observable.observable")
+
+    scheduler_module = types.ModuleType("rx.scheduler")
+    scheduler_module.__path__ = []
+    eventloop_module = types.ModuleType("rx.scheduler.eventloop")
+
+    subject_module = types.ModuleType("rx.subject")
+    subject_module.__path__ = []
+    subject_behaviorsubject_module = types.ModuleType("rx.subject.behaviorsubject")
+    subject_behaviorsubject_module.__path__ = []
+    subject_subject_module = types.ModuleType("rx.subject.subject")
+    subject_subject_module.__path__ = []
+
+    operators_module = types.ModuleType("rx.operators")
+
+    class Observable:  # pragma: no cover - stub placeholder
+        def subscribe(self, *args, **kwargs):
+            return MagicMock()
+
+    class Observer:  # pragma: no cover - stub placeholder
+        def on_next(self, value):
+            pass
+
+        def on_completed(self):
+            pass
+
+        def on_error(self, error):
+            pass
+
+    class AsyncIOScheduler:  # pragma: no cover - stub placeholder
+        def __init__(self, *args, **kwargs):
+            pass
+
+    class _BaseSubject:
+        def __init__(self, *args, **kwargs):
+            self._value = kwargs.get("value")
+
+        def on_next(self, value):
+            self._value = value
+
+        def subscribe(self, *args, **kwargs):
+            return MagicMock()
+
+    class Subject(_BaseSubject):
+        pass
+
+    class BehaviorSubject(_BaseSubject):
+        pass
+
+    def _identity_operator(*args, **kwargs):
+        def _apply(observable):
+            return observable
+
+        return _apply
+
+    core_module.Observable = Observable
+    core_module.Observer = Observer
+    core_observable_module.Observable = Observable
+    core_observable_submodule.Observable = Observable
+
+    subject_module.Subject = Subject
+    subject_module.BehaviorSubject = BehaviorSubject
+    subject_behaviorsubject_module.BehaviorSubject = BehaviorSubject
+    subject_subject_module.Subject = Subject
+
+    eventloop_module.AsyncIOScheduler = AsyncIOScheduler
+    scheduler_module.eventloop = eventloop_module
+
+    operators_module.map = _identity_operator
+    operators_module.do = _identity_operator
+
+    rx_module.core = core_module
+    rx_module.scheduler = scheduler_module
+    rx_module.subject = subject_module
+    rx_module.operators = operators_module
+
+    sys.modules["rx"] = rx_module
+    sys.modules["rx.core"] = core_module
+    sys.modules["rx.core.observable"] = core_observable_module
+    sys.modules["rx.core.observable.observable"] = core_observable_submodule
+    sys.modules["rx.scheduler"] = scheduler_module
+    sys.modules["rx.scheduler.eventloop"] = eventloop_module
+    sys.modules["rx.subject"] = subject_module
+    sys.modules["rx.subject.behaviorsubject"] = subject_behaviorsubject_module
+    sys.modules["rx.subject.subject"] = subject_subject_module
+    sys.modules["rx.operators"] = operators_module
+
+if "sqlalchemy" not in sys.modules:
+    sqlalchemy_module = types.ModuleType("sqlalchemy")
+    exc_module = types.ModuleType("sqlalchemy.exc")
+
+    class SQLAlchemyError(Exception):
+        pass
+
+    exc_module.SQLAlchemyError = SQLAlchemyError
+    sqlalchemy_module.exc = exc_module
+
+    sys.modules["sqlalchemy"] = sqlalchemy_module
+    sys.modules["sqlalchemy.exc"] = exc_module
+
 from typer.testing import CliRunner  # noqa: E402
 
 from cleveragents.cli.commands.action import app as action_app  # noqa: E402
@@ -400,6 +603,118 @@ def plan_list_namespace_short() -> None:
         print("cli-lifecycle-plan-list-namespace-short-ok")
 
 
+def plan_execute_rich_panels() -> None:
+    """Verify rich plan execute output renders spec panels with fallbacks."""
+
+    mock_service = MagicMock()
+
+    plan_before = _mock_plan(
+        phase=PlanPhase.STRATEGIZE,
+        state=ProcessingState.COMPLETE,
+        plan_id=_PLAN_ULID,
+    )
+    plan_queued = _mock_plan(
+        phase=PlanPhase.EXECUTE,
+        state=ProcessingState.QUEUED,
+        plan_id=_PLAN_ULID,
+    )
+    plan_complete = _mock_plan(
+        phase=PlanPhase.EXECUTE,
+        state=ProcessingState.COMPLETE,
+        plan_id=_PLAN_ULID,
+    )
+    plan_complete.identity = PlanIdentity(plan_id=_PLAN_ULID, attempt=3)
+    plan_complete.timestamps.execute_started_at = datetime(2026, 4, 10, 12, 34, 56)
+    plan_complete.sandbox_refs = ["/tmp/worktrees/plan"]
+
+    mock_service.get_plan.side_effect = [plan_before, plan_queued, plan_complete]
+    mock_service.execute_plan.return_value = plan_queued
+
+    envelope = {
+        "command": "plan execute",
+        "status": "ok",
+        "exit_code": 0,
+        "data": {
+            "plan_id": _PLAN_ULID,
+            "phase": "execute",
+            "sandbox": {
+                "strategy": "git_worktree",
+                "path": "/tmp/worktrees/plan",
+                "branch": "cleveragents/plan-01KHDE6W",
+                "status": "active",
+            },
+            "worker": "local/executor",
+            "strategy_summary": {
+                "decisions": 4,
+                "invariants": 1,
+                "planned_child_plans": 2,
+                "estimated_files": 12,
+                "risk": "low",
+            },
+            "progress": [
+                {"label": "Collect context", "status": "complete"},
+                {"label": "Run tools", "status": "running"},
+                {"label": "Build changeset", "status": "pending"},
+                {"label": "Validate", "status": "pending"},
+            ],
+        },
+        "messages": ["Execution started"],
+    }
+
+    with (
+        patch(
+            "cleveragents.cli.commands.plan._get_lifecycle_service",
+            return_value=mock_service,
+        ),
+        patch(
+            "cleveragents.cli.commands.plan._get_plan_executor",
+            return_value=MagicMock(),
+        ),
+        patch(
+            "cleveragents.cli.commands.plan._create_sandbox_for_plan",
+            return_value=(None, None),
+        ),
+        patch(
+            "cleveragents.cli.commands.plan._execute_output_dict",
+            return_value=envelope,
+        ),
+    ):
+        result = runner.invoke(plan_app, ["execute", _PLAN_ULID])
+
+    if result.exit_code != 0:
+        print(
+            f"FAIL: plan execute rich panels returned {result.exit_code}",
+            file=sys.stderr,
+        )
+        print(result.output, file=sys.stderr)
+        sys.exit(1)
+
+    sanitized = re.sub(r"\x1b\[[0-9;]*m", "", result.output)
+
+    expected_fragments = [
+        "Execution",
+        "Sandbox",
+        "Strategy Summary",
+        "Progress",
+        "Started:",
+        "12:34:56",
+        "Attempt:",
+        "3",
+        "✓ OK Execution started",
+    ]
+
+    for fragment in expected_fragments:
+        if fragment not in sanitized:
+            print(
+                f"FAIL: Expected rich output fragment '{fragment}' not found",
+                file=sys.stderr,
+            )
+            print(sanitized, file=sys.stderr)
+            sys.exit(1)
+
+    print("cli-lifecycle-plan-execute-rich-panels-ok")
+
+
 # ---------------------------------------------------------------------------
 # Main dispatcher
 # ---------------------------------------------------------------------------
@@ -415,6 +730,7 @@ _COMMANDS = {
     "plan-list-namespace": plan_list_namespace,
     "plan-list-namespace-short": plan_list_namespace_short,
     "full-lifecycle": full_lifecycle,
+    "plan-execute-rich-panels": plan_execute_rich_panels,
 }
 
 

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,174 @@ 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
+
+    timestamps = getattr(plan, "timestamps", None)
+    fallback_started_at = getattr(timestamps, "execute_started_at", None)
+    started_display: str
+    if isinstance(started_value, str) and started_value.strip():
+        started_display = started_value
+    elif isinstance(fallback_started_at, datetime):
+        started_display = fallback_started_at.strftime("%H:%M:%S")
+    else:
+        started_display = "—"
+
+    identity = getattr(plan, "identity", None)
+    fallback_attempt = getattr(identity, "attempt", None)
+    attempt_display: str
+    if isinstance(attempt_value, int):
+        attempt_display = str(attempt_value)
+    elif isinstance(attempt_value, str) and attempt_value.strip():
+        attempt_display = attempt_value.strip()
+    elif isinstance(fallback_attempt, int) and fallback_attempt >= 1:
+        attempt_display = str(fallback_attempt)
+    else:
+        attempt_display = "1"
+
+    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),
+        ("[green bold]Started:[/green bold]", started_display),
+        ("[#5599ff bold]Attempt:[/#5599ff bold]", attempt_display),
+    ]
+
+    _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 +2619,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 +4474,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 +4486,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