cleveragents-core/robot/helper_plan_subplan_tool.py

"""Helper script for Robot Framework plan_subplan tool smoke tests.

Each command prints a sentinel string on success so the Robot suite can
assert with ``Should Contain``.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from cleveragents.domain.models.core.decision import Decision, DecisionType
from cleveragents.tool.builtins import register_subplan_tool
from cleveragents.tool.builtins.subplan_tool import (
    PLAN_SUBPLAN_SPEC,
    SubplanPayload,
    make_plan_subplan_spec,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner

_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"


# ---------------------------------------------------------------------------
# Stub decision service
# ---------------------------------------------------------------------------


class _StubService:
    def __init__(self) -> None:
        self.recorded: list[Decision] = []

    def record_decision(self, decision: Decision) -> Decision:
        self.recorded.append(decision)
        return decision


# ---------------------------------------------------------------------------
# Helper to build minimal valid inputs
# ---------------------------------------------------------------------------


def _base(
    goal: str = "Refactor module",
    resource_scopes: list[str] | None = None,
    parallel: bool = False,
    **extra: Any,
) -> dict[str, Any]:
    return {
        "goal": goal,
        "plan_id": _PLAN_ID,
        "sequence_number": 0,
        "resource_scopes": resource_scopes or ["src/"],
        "parallel": parallel,
        **extra,
    }


# ---------------------------------------------------------------------------
# Test functions
# ---------------------------------------------------------------------------


def test_serial_spawn() -> None:
    result = PLAN_SUBPLAN_SPEC.handler(
        _base(goal="Refactor auth", resource_scopes=["src/auth/"])
    )
    assert result["success"] is True
    assert result["decision_type"] == "subplan_spawn"
    assert result["parallel"] is False
    print("serial-spawn-ok")


def test_parallel_spawn() -> None:
    result = PLAN_SUBPLAN_SPEC.handler(
        _base(
            goal="Run tests in parallel",
            resource_scopes=["tests/"],
            parallel=True,
            max_parallel=3,
            dependencies=[],
        )
    )
    assert result["success"] is True
    assert result["decision_type"] == "subplan_parallel_spawn"
    assert result["parallel"] is True
    print("parallel-spawn-ok")


def test_validation_error() -> None:
    result = PLAN_SUBPLAN_SPEC.handler(
        {"goal": "Bad", "plan_id": _PLAN_ID, "sequence_number": 0}
    )
    assert result["success"] is False
    err = result["error"].lower()
    assert "resource_scopes" in err or "project_ref" in err
    print("validation-error-ok")


def test_defaults() -> None:
    result = PLAN_SUBPLAN_SPEC.handler(_base())
    assert result["merge_strategy"] == "git_three_way"
    assert result["max_parallel"] == 5
    assert result["dependencies"] == []
    print("defaults-ok")


def test_decision_service_injection() -> None:
    stub = _StubService()
    spec = make_plan_subplan_spec(decision_service=stub)
    result = spec.handler(_base(goal="Run linting", resource_scopes=["src/"]))
    assert result["success"] is True
    assert len(stub.recorded) == 1
    assert stub.recorded[0].decision_type == DecisionType.SUBPLAN_SPAWN
    print("service-injection-ok")


def test_registration() -> None:
    registry = ToolRegistry()
    register_subplan_tool(registry)
    spec = registry.get("builtin/plan-subplan")
    assert spec is not None
    assert spec.name == "builtin/plan-subplan"
    print("registration-ok")


def test_runner_execution() -> None:
    registry = ToolRegistry()
    register_subplan_tool(registry)
    runner = ToolRunner(registry)
    result = runner.execute(
        "builtin/plan-subplan",
        _base(goal="Via runner", resource_scopes=["src/"]),
    )
    assert result.success, f"Runner failed: {result.error}"
    assert result.output["decision_type"] == "subplan_spawn"
    print("runner-execution-ok")


def test_rationale_content() -> None:
    result = PLAN_SUBPLAN_SPEC.handler(
        _base(goal="Migrate to PostgreSQL", resource_scopes=["db/"])
    )
    assert result["success"] is True
    rationale = result["rationale"]
    assert "Migrate to PostgreSQL" in rationale
    assert "sequential" in rationale
    print("rationale-ok")


def test_payload_model_direct() -> None:
    payload = SubplanPayload(
        goal="Direct model test",
        plan_id=_PLAN_ID,
        sequence_number=1,
        resource_scopes=["src/"],
        parallel=True,
        max_parallel=10,
    )
    assert payload.merge_strategy.value == "git_three_way"
    assert payload.max_parallel == 10
    assert payload.parallel is True
    print("payload-model-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Any] = {
    "serial": test_serial_spawn,
    "parallel": test_parallel_spawn,
    "validation": test_validation_error,
    "defaults": test_defaults,
    "service": test_decision_service_injection,
    "registration": test_registration,
    "runner": test_runner_execution,
    "rationale": test_rationale_content,
    "payload": test_payload_model_direct,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()