cleveragents-core/robot/helper_wf03_plan_prompt_confidence.py

"""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())