diff --git a/features/plan_prompt_command.feature b/features/plan_prompt_command.feature new file mode 100644 index 000000000..cd112c134 --- /dev/null +++ b/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 "" + 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" diff --git a/features/steps/plan_prompt_command_steps.py b/features/steps/plan_prompt_command_steps.py new file mode 100644 index 000000000..97cf2f828 --- /dev/null +++ b/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 diff --git a/noxfile.py b/noxfile.py index cb3fca9c4..c1e69a2b0 100644 --- a/noxfile.py +++ b/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}", diff --git a/src/cleveragents/a2a/facade.py b/src/cleveragents/a2a/facade.py index 8fdebfca4..1fb3536ad 100644 --- a/src/cleveragents/a2a/facade.py +++ b/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, diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 84572bc9b..ab562608b 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/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. diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 610ebc1bc..97ff2bfd9 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/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, diff --git a/src/cleveragents/tool/wrapping.py b/src/cleveragents/tool/wrapping.py index 0566a3a10..a468c83b3 100644 --- a/src/cleveragents/tool/wrapping.py +++ b/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"", "exec") + self._compiled = compile( # nosemgrep: no-compile-exec + self._code, + f"", + "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,