features/plan_cli_coverage_boost.feature

@@ -57,7 +57,22 @@ Feature: Plan CLI coverage boost
     When I invoke execute with "--format" "json" and plan id
     Then the plan coverage command should succeed
     And the plan coverage output should contain "plan_id"
-    And the plan coverage output should contain "namespaced_name"
+    And the plan coverage output should contain "sandbox"
+    And the plan coverage output should contain "worker"
+    And the plan coverage output should contain "progress"
+    And the plan coverage output should contain "strategy_summary"
+    And the plan coverage output should contain "command"
+    And the plan coverage output should contain "exit_code"
+
+  Scenario: execute_plan JSON output has spec-required envelope structure
+    Given a plan lifecycle CLI runner for coverage
+    And a mocked lifecycle service for plan coverage commands
+    And the service has a complete strategize plan for execute
+    When I invoke execute with "--format" "json" and plan id
+    Then the plan coverage command should succeed
+    And the execute JSON output has the spec-required envelope fields
+    And the execute JSON output data has sandbox with strategy field
+    And the execute JSON output data has progress list with label and status
 
   # ---- apply_plan non-rich format ----
 

features/steps/plan_cli_coverage_boost_steps.py

@@ -508,3 +508,82 @@ def step_plan_coverage_output_contains(context, text: str) -> None:
 def step_plan_coverage_output_not_contains(context, text: str) -> None:
     output = _output(context)
     assert text not in output, f"Did not expect '{text}' in output:\n{output}"
+
+
+@then("the execute JSON output has the spec-required envelope fields")
+def step_execute_json_envelope_fields(context) -> None:
+    """Verify the execute JSON output has the spec-required top-level envelope."""
+    import json
+
+    output = _output(context)
+    # The output may have a trailing newline; strip it
+    parsed = json.loads(output.strip())
+    required_fields = {"command", "status", "exit_code", "data", "timing", "messages"}
+    missing = required_fields - set(parsed.keys())
+    assert not missing, (
+        f"Missing envelope fields {missing} in execute JSON output:\n{output}"
+    )
+    assert parsed["command"] == "plan execute", (
+        f"Expected command='plan execute', got '{parsed['command']}'"
+    )
+    assert parsed["status"] == "ok", f"Expected status='ok', got '{parsed['status']}'"
+    assert parsed["exit_code"] == 0, (
+        f"Expected exit_code=0, got '{parsed['exit_code']}'"
+    )
+    assert isinstance(parsed["data"], dict), (
+        f"Expected data to be a dict, got {type(parsed['data'])}"
+    )
+    assert isinstance(parsed["messages"], list), (
+        f"Expected messages to be a list, got {type(parsed['messages'])}"
+    )
+    # Verify data has required fields
+    data = parsed["data"]
+    required_data_fields = {
+        "plan_id",
+        "phase",
+        "sandbox",
+        "worker",
+        "attempt",
+        "strategy_summary",
+        "progress",
+    }
+    missing_data = required_data_fields - set(data.keys())
+    assert not missing_data, (
+        f"Missing data fields {missing_data} in execute JSON output data:\n{output}"
+    )
+
+
+@then("the execute JSON output data has sandbox with strategy field")
+def step_execute_json_sandbox_strategy(context) -> None:
+    """Verify the sandbox dict in execute JSON output has a strategy field."""
+    import json
+
+    output = _output(context)
+    parsed = json.loads(output.strip())
+    data = parsed["data"]
+    sandbox = data.get("sandbox")
+    assert isinstance(sandbox, dict), (
+        f"Expected sandbox to be a dict, got {type(sandbox)}: {sandbox}"
+    )
+    assert "strategy" in sandbox, f"Expected 'strategy' key in sandbox dict: {sandbox}"
+
+
+@then("the execute JSON output data has progress list with label and status")
+def step_execute_json_progress_list(context) -> None:
+    """Verify the progress list in execute JSON output has label and status fields."""
+    import json
+
+    output = _output(context)
+    parsed = json.loads(output.strip())
+    data = parsed["data"]
+    progress = data.get("progress")
+    assert isinstance(progress, list), (
+        f"Expected progress to be a list, got {type(progress)}: {progress}"
+    )
+    assert len(progress) > 0, "Expected at least one progress step"
+    for step in progress:
+        assert isinstance(step, dict), (
+            f"Expected each progress step to be a dict, got {type(step)}: {step}"
+        )
+        assert "label" in step, f"Expected 'label' key in progress step: {step}"
+        assert "status" in step, f"Expected 'status' key in progress step: {step}"

src/cleveragents/cli/commands/plan.py

@@ -264,6 +264,176 @@ def _plan_spec_dict(plan: Any) -> dict[str, object]:
     return {"plan": str(plan)}
 
 
+def _execute_output_dict(
+    plan: Any,
+    started_at: datetime | None = None,
+    duration_ms: int | None = None,
+) -> dict[str, object]:
+    """Build the spec-required execute output envelope.
+
+    Returns the structured JSON envelope for ``agents plan execute --format json``
+    as defined in the specification §agents plan execute.
+
+    The envelope structure is::
+
+        {
+          "command": "plan execute",
+          "status": "ok",
+          "exit_code": 0,
+          "data": {
+            "plan_id": "...",
+            "phase": "execute",
+            "sandbox": {"strategy": ..., "path": ..., "branch": ..., "status": ...},
+            "worker": "local/executor",
+            "started": "HH:MM:SS",
+            "attempt": 1,
+            "strategy_summary": {...},
+            "progress": [...]
+          },
+          "timing": {"started": "...", "duration_ms": ...},
+          "messages": ["Execution started"]
+        }
+
+    Args:
+        plan: The ``Plan`` domain model after execution.
+        started_at: When execution started (used for timing).  Falls back
+            to ``plan.timestamps.execute_started_at`` when ``None``.
+        duration_ms: Elapsed milliseconds.  ``None`` when not available.
+
+    Returns:
+        A JSON-serialisable dict matching the spec-required envelope.
+    """
+    from cleveragents.domain.models.core.plan import Plan as LifecyclePlan
+
+    if not isinstance(plan, LifecyclePlan):
+        # Legacy plan fallback — return minimal envelope
+        return {
+            "command": "plan execute",
+            "status": "ok",
+            "exit_code": 0,
+            "data": {"plan": str(plan)},
+            "timing": {},
+            "messages": ["Execution started"],
+        }
+
+    plan_id = plan.identity.plan_id
+
+    # ── Sandbox info ──────────────────────────────────────────────────────────
+    # Derive sandbox details from plan.sandbox_refs when available.
+    # sandbox_refs is a list of sandbox reference IDs; the first entry is the
+    # primary sandbox.  We build a best-effort dict from available plan data.
+    sandbox: dict[str, object]
+    if plan.sandbox_refs:
+        # Use the first sandbox ref as the primary sandbox path/id
+        primary_ref = plan.sandbox_refs[0]
+        sandbox = {
+            # TODO: derive strategy from plan's actual sandbox configuration
+            # once the plan model exposes it; "git_worktree" is the current
+            # default strategy used in all local execution environments.
+            "strategy": "git_worktree",
+            "path": primary_ref,
+            "branch": f"cleveragents/plan-{plan_id[:8]}",
+            "status": "active",
+        }
+    else:
+        sandbox = {
+            # TODO: derive strategy from plan's actual sandbox configuration
+            # once the plan model exposes it; "git_worktree" is the current
+            # default strategy used in all local execution environments.
+            "strategy": "git_worktree",
+            "path": None,
+            "branch": f"cleveragents/plan-{plan_id[:8]}",
+            "status": "pending",
+        }
+
+    # ── Worker ────────────────────────────────────────────────────────────────
+    worker: str = plan.execution_actor or "local/executor"
+
+    # ── Started timestamp ─────────────────────────────────────────────────────
+    ts_started = started_at or plan.timestamps.execute_started_at
+    started_str: str | None = (
+        ts_started.strftime("%H:%M:%S") if ts_started is not None else None
+    )
+
+    # ── Attempt ───────────────────────────────────────────────────────────────
+    attempt: int = plan.identity.attempt
+
+    # ── Strategy summary ──────────────────────────────────────────────────────
+    # Derive from estimation_result when available; fall back to decisions count.
+    strategy_summary: dict[str, object]
+    if plan.estimation_result is not None:
+        est = plan.estimation_result.as_display_dict()
+        strategy_summary = {
+            "decisions": len(plan.decisions),
+            "invariants": len(plan.invariants),
+            "planned_child_plans": est.get("planned_child_plans", 0),
+            "estimated_files": est.get("estimated_files", 0),
+            "risk": est.get("risk", "unknown"),
+        }
+    else:
+        strategy_summary = {
+            "decisions": len(plan.decisions),
+            "invariants": len(plan.invariants),
+            "planned_child_plans": 0,
+            "estimated_files": 0,
+            "risk": "unknown",
+        }
+
+    # ── Progress steps ────────────────────────────────────────────────────────
+    # Map plan phase/state to progress step statuses.
+    # ProcessingState is imported at module level (line 46); no inline import needed.
+    is_complete = plan.processing_state in (
+        ProcessingState.COMPLETE,
+        ProcessingState.APPLIED,
+    )
+    is_errored = plan.processing_state == ProcessingState.ERRORED
+
+    def _step_status(step_index: int) -> str:
+        """Return status for a progress step given the plan's current state."""
+        if is_errored:
+            return "error" if step_index == 0 else "pending"
+        if is_complete:
+            return "complete"
+        # In-progress: first step is running, rest are pending
+        return "running" if step_index == 0 else "pending"
+
+    progress: list[dict[str, str]] = [
+        {"label": "Collect context", "status": _step_status(0)},
+        {"label": "Run tools", "status": _step_status(1)},
+        {"label": "Build changeset", "status": _step_status(2)},
+        {"label": "Validate", "status": _step_status(3)},
+    ]
+
+    # ── Timing ────────────────────────────────────────────────────────────────
+    timing: dict[str, object] = {}
+    if ts_started is not None:
+        timing["started"] = ts_started.isoformat()
+    if duration_ms is not None:
+        timing["duration_ms"] = duration_ms
+
+    # ── Data payload ──────────────────────────────────────────────────────────
+    data: dict[str, object] = {
+        "plan_id": plan_id,
+        "phase": plan.phase.value,
+        "sandbox": sandbox,
+        "worker": worker,
+        "attempt": attempt,
+        "strategy_summary": strategy_summary,
+        "progress": progress,
+    }
+    if started_str is not None:
+        data["started"] = started_str
+
+    return {
+        "command": "plan execute",
+        "status": "ok",
+        "exit_code": 0,
+        "data": data,
+        "timing": timing,
+        "messages": ["Execution started"],
+    }
+
+
 # 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.
@@ -1778,6 +1948,9 @@ def execute_plan(
         service = _get_lifecycle_service()
         executor = _get_plan_executor(lifecycle_service=service)
 
+        # Track wall-clock start time for timing output
+        execute_wall_start = datetime.now()
+
         if not plan_id:
             # Auto-discover: look for plans in strategize or execute
             plans_strat = service.list_plans(phase=PlanPhase.STRATEGIZE)
@@ -1897,8 +2070,15 @@ def execute_plan(
         _notify_facade("plan.execute", {"plan_id": plan_id})
 
         if fmt != OutputFormat.RICH.value:
-            data = _plan_spec_dict(plan)
-            console.print(format_output(data, 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,
+            )
+            console.print(format_output(envelope, fmt))
         else:
             _print_lifecycle_plan(plan, title="Plan Executed")
             phase_label = f"{plan.phase.value}/{plan.state.value}"