"""Robot Framework helper -- WF03 plan prompt & confidence-threshold pausing. Exercises: - plan prompt via the A2A facade (S15822) - confidence-threshold pausing with the cautious profile (S37262-37367) - pause-and-resume flow: low confidence -> pause -> plan prompt -> resume """ # ruff: noqa: E402 from __future__ import annotations import sys from collections.abc import Callable from pathlib import Path from typing import NoReturn _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) _ROBOT = str(Path(__file__).resolve().parent) if _ROBOT not in sys.path: sys.path.insert(0, _ROBOT) from helpers_common import reset_global_state from cleveragents.a2a.facade import A2aLocalFacade from cleveragents.a2a.models import A2aRequest from cleveragents.application.services.autonomy_controller import ( AutonomyController, ) from cleveragents.domain.models.core.automation_profile import ( BUILTIN_PROFILES, ) from cleveragents.domain.models.core.escalation import ( ConfidenceFactors, OperationContext, ) def _fail(msg: str) -> NoReturn: print(f"FAIL: {msg}", file=sys.stderr) raise SystemExit(1) def _factors_for_confidence(target: float) -> ConfidenceFactors: """Build factors that produce a specific confidence score. Uses the AutonomyController formula with default weights: score = 0.30*psr + 0.20*cf + 0.30*(1-risk) + 0.20*(1-ic) Setting psr=T, cf=T, risk=(1-T), ic=(1-T) yields score=T. """ return ConfidenceFactors( past_success_rate=target, codebase_familiarity=target, risk_assessment=1.0 - target, invariant_complexity=1.0 - target, ) # --------------------------------------------------------------------------- # Test: plan prompt via A2A facade # --------------------------------------------------------------------------- def wf03_plan_prompt_facade() -> None: """Exercise plan prompt via the A2A facade dispatch path.""" facade = A2aLocalFacade() plan_id = "PLAN-WF03-PROMPT-001" guidance = "Use separate models/ directory instead of inline" request = A2aRequest( method="_cleveragents/plan/prompt", params={"plan_id": plan_id, "guidance": guidance}, ) response = facade.dispatch(request) if response.result is None: _fail(f"Expected result, got error: {response.error}") data = response.result if data.get("status") != "guidance_injected": _fail(f"Expected data.status 'guidance_injected', got: {data}") if data.get("plan_id") != plan_id: _fail(f"Expected plan_id '{plan_id}', got: {data}") # Verify guidance propagates through the dispatch path actual_guidance = data.get("guidance") if actual_guidance != guidance: _fail(f"Expected guidance '{guidance}', got: {actual_guidance}") # Verify the stub flag is set (current implementation is a stub) if not data.get("stub"): _fail(f"Expected stub=True, got: {data}") # TODO(#885): Verify guidance_added.scope, state_transition, # decision_created.type per S15894-15913 print("wf03-plan-prompt-ok") # --------------------------------------------------------------------------- # Test: confidence-threshold pausing # --------------------------------------------------------------------------- def wf03_confidence_threshold_pausing() -> None: """Verify cautious profile pauses on low-confidence decisions. The cautious profile has edit_code=0.6. With confidence 0.55, the system should NOT proceed (proceed=False). """ controller = AutonomyController() profile = BUILTIN_PROFILES["cautious"] # Verify the cautious threshold value threshold = profile.edit_code if abs(threshold - 0.6) > 1e-6: _fail(f"Expected cautious edit_code=0.6, got {threshold}") # Produce confidence 0.55 (below threshold 0.6) factors = _factors_for_confidence(0.55) operation = OperationContext( operation_type="edit_code", ) decision = controller.should_proceed_automatically( operation=operation, factors=factors, profile=profile ) # Verify confidence is 0.55 if abs(decision.confidence - 0.55) > 1e-6: _fail(f"Expected confidence 0.55, got {decision.confidence}") # Verify the system pauses (does NOT proceed) if decision.proceed is not False: _fail( "Expected proceed=False (pause) at " f"confidence={decision.confidence:.3f} " f"< threshold={decision.threshold:.3f}, " "but got proceed=True" ) # Verify threshold is 0.6 if abs(decision.threshold - 0.6) > 1e-6: _fail(f"Expected threshold 0.6, got {decision.threshold}") # Verify explanation mentions the threshold if "threshold" not in decision.explanation.lower(): _fail(f"Explanation should mention threshold: {decision.explanation}") print("wf03-confidence-pause-ok") # --------------------------------------------------------------------------- # Test: confidence-threshold proceeds # --------------------------------------------------------------------------- def wf03_confidence_threshold_proceeds() -> None: """Verify cautious profile proceeds when confidence >= 0.6. For strategy decisions, the cautious profile should proceed. """ controller = AutonomyController() profile = BUILTIN_PROFILES["cautious"] # Produce confidence 0.85 (above threshold 0.6) factors = _factors_for_confidence(0.85) operation = OperationContext( operation_type="edit_code", ) decision = controller.should_proceed_automatically( operation=operation, factors=factors, profile=profile ) # Verify confidence is 0.85 if abs(decision.confidence - 0.85) > 1e-6: _fail(f"Expected confidence 0.85, got {decision.confidence}") # Verify the system proceeds if decision.proceed is not True: _fail( "Expected proceed=True at " f"confidence={decision.confidence:.3f} " f">= threshold={decision.threshold:.3f}, " "but got proceed=False" ) # Also test exactly at threshold (0.6 should proceed) factors_exact = _factors_for_confidence(0.6) decision_exact = controller.should_proceed_automatically( operation=operation, factors=factors_exact, profile=profile ) if decision_exact.proceed is not True: _fail( "Expected proceed=True at exactly threshold, " f"confidence={decision_exact.confidence:.3f} " f">= threshold={decision_exact.threshold:.3f}" ) print("wf03-confidence-proceed-ok") # --------------------------------------------------------------------------- # Test: pause-and-resume via plan prompt # --------------------------------------------------------------------------- def wf03_pause_and_resume() -> None: """End-to-end: low confidence -> pause -> plan prompt -> resume. Simulates the WF03 flow from spec S37262-37367: 1. A strategy decision has confidence 0.55 (below cautious 0.6) 2. The system pauses (proceed=False) 3. User provides guidance via plan prompt (plan_id from context) 4. The facade accepts the guidance and echoes it back 5. Re-evaluation with higher confidence proceeds NOTE: Full causal resume testing (guidance -> re-evaluation) requires the wired actor/provider stack from #885. This test establishes a logical connection via plan_id but cannot prove causality yet. """ # Step 1: Verify the system would pause controller = AutonomyController() profile = BUILTIN_PROFILES["cautious"] factors = _factors_for_confidence(0.55) operation = OperationContext( operation_type="edit_code", ) decision = controller.should_proceed_automatically( operation=operation, factors=factors, profile=profile ) if decision.proceed is not False: _fail( "Step 1: Expected pause (proceed=False) at confidence " f"{decision.confidence:.3f}, got proceed=True" ) # Step 2: Verify explanation indicates escalation explanation_lower = decision.explanation.lower() if not any( w in explanation_lower for w in ("escalat", "user", "manual", "human", "below") ): _fail(f"Step 2: Explanation should indicate escalation: {decision.explanation}") # Step 3: Provide guidance via plan prompt, using a plan_id that # logically connects to the paused decision context paused_plan_id = "PLAN-WF03-PAUSE-RESUME" facade = A2aLocalFacade() guidance_text = ( "Use separate models/ directory with one model per file. " "This follows the project convention." ) prompt_request = A2aRequest( method="_cleveragents/plan/prompt", params={ "plan_id": paused_plan_id, "guidance": guidance_text, }, ) prompt_response = facade.dispatch(prompt_request) if prompt_response.result is None: _fail(f"Step 3: plan prompt failed: {prompt_response}") if prompt_response.result.get("status") != "guidance_injected": _fail(f"Step 3: Expected guidance_injected: {prompt_response.result}") # Verify guidance propagates through the dispatch path actual_guidance = prompt_response.result.get("guidance") if actual_guidance != guidance_text: _fail(f"Step 3: Expected guidance '{guidance_text}', got: {actual_guidance}") # Verify the plan_id from response matches paused context actual_plan_id = prompt_response.result.get("plan_id") if actual_plan_id != paused_plan_id: _fail(f"Step 3: Expected plan_id '{paused_plan_id}', got: {actual_plan_id}") # Step 4: Verify the system would proceed after correction corrected_factors = _factors_for_confidence(0.85) corrected_decision = controller.should_proceed_automatically( operation=operation, factors=corrected_factors, profile=profile, ) if corrected_decision.proceed is not True: _fail( "Step 4: Expected proceed=True after correction with " f"confidence {corrected_decision.confidence:.3f}, " "got proceed=False" ) print("wf03-pause-resume-ok") # --------------------------------------------------------------------------- # Dispatcher # --------------------------------------------------------------------------- _COMMANDS: dict[str, Callable[[], None]] = { "plan-prompt-facade": wf03_plan_prompt_facade, "confidence-threshold-pausing": wf03_confidence_threshold_pausing, "confidence-threshold-proceeds": wf03_confidence_threshold_proceeds, "pause-and-resume": wf03_pause_and_resume, } def main() -> int: """Entry point called by Robot Framework ``Run Process``.""" if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: valid = "|".join(_COMMANDS) print( f"Usage: helper_wf03_plan_prompt_confidence.py <{valid}>", file=sys.stderr, ) return 1 reset_global_state() _COMMANDS[sys.argv[1]]() return 0 if __name__ == "__main__": sys.exit(main())