cleveragents-core/robot/helper_subplan_spawn.py

"""Helper script for subplan spawn Robot Framework smoke tests.

Exercises SubplanService spawn workflow, validation, and metadata
persistence without requiring the full service layer.
"""

from __future__ import annotations

import sys
from unittest.mock import MagicMock

from cleveragents.application.services.subplan_service import (
    SpawnEntry,
    SubplanService,
)
from cleveragents.domain.models.core.decision import (
    ContextSnapshot,
    Decision,
    DecisionType,
    ResourceRef,
)
from cleveragents.domain.models.core.plan import (
    ExecutionMode,
    NamespacedName,
    Plan,
    PlanIdentity,
    SubplanConfig,
)

_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
_ROOT_ID = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
_DEC_ID1 = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
_DEC_ID2 = "01HGZ6FE0AQDYTR4BXVQZ6DB00"
_DEC_ID3 = "01HGZ6FE0AQDYTR4BXVQZ6DC00"


def _mock_decision_service() -> MagicMock:
    svc: MagicMock = MagicMock()
    svc.list_by_type = MagicMock(return_value=[])
    return svc


def _make_decision(
    decision_id: str,
    plan_id: str = _PLAN_ID,
    decision_type: DecisionType = DecisionType.SUBPLAN_SPAWN,
    sequence: int = 0,
    chosen_option: str = "local/sub-action",
    resource_ids: list[str] | None = None,
) -> Decision:
    resources: list[ResourceRef] = []
    if resource_ids:
        resources = [ResourceRef(resource_id=rid) for rid in resource_ids]
    return Decision(
        decision_id=decision_id,
        plan_id=plan_id,
        decision_type=decision_type,
        sequence_number=sequence,
        question="Spawn child plan?",
        chosen_option=chosen_option,
        context_snapshot=ContextSnapshot(relevant_resources=resources),
    )


def _make_plan(
    plan_id: str = _PLAN_ID,
    parent_plan_id: str | None = None,
    root_plan_id: str | None = None,
) -> Plan:
    return Plan(
        identity=PlanIdentity(
            plan_id=plan_id,
            parent_plan_id=parent_plan_id,
            root_plan_id=root_plan_id,
        ),
        namespaced_name=NamespacedName(namespace="local", name="test-plan"),
        description="Robot test plan",
        action_name="local/test-action",
    )


def _spawn_single() -> None:
    """Spawn a single child plan from a decision entry."""
    svc = SubplanService(decision_service=_mock_decision_service())
    plan = _make_plan()
    config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
    dec = _make_decision(_DEC_ID1)
    entries = [SpawnEntry(decision=dec, action_name="local/sub-action")]
    result = svc.spawn(parent_plan=plan, config=config, spawn_entries=entries)
    assert result.total_spawned == 1
    assert len(result.spawned_statuses) == 1
    meta = next(iter(result.metadata.values()))
    assert meta.parent_plan_id == _PLAN_ID
    print("spawn-single-ok")


def _spawn_multiple() -> None:
    """Spawn multiple child plans from decision entries."""
    svc = SubplanService(decision_service=_mock_decision_service())
    plan = _make_plan()
    config = SubplanConfig(
        execution_mode=ExecutionMode.PARALLEL,
        max_parallel=5,
    )
    entries = []
    for i, did in enumerate([_DEC_ID1, _DEC_ID2, _DEC_ID3]):
        dec = _make_decision(did, sequence=i)
        entries.append(SpawnEntry(decision=dec, action_name=f"local/sub-{i}"))
    result = svc.spawn(parent_plan=plan, config=config, spawn_entries=entries)
    assert result.total_spawned == 3
    ids = [s.subplan_id for s in result.spawned_statuses]
    assert len(ids) == len(set(ids)), "Duplicate IDs"
    print("spawn-multiple-ok")


def _validate_resource_scope() -> None:
    """Validate spawn with missing resource scope."""
    svc = SubplanService(decision_service=_mock_decision_service())
    config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
    dec = _make_decision(_DEC_ID1, resource_ids=["res-missing"])
    entries = [
        SpawnEntry(
            decision=dec,
            action_name="local/sub-action",
            target_resources=["res-missing"],
        )
    ]
    result = svc.validate_spawn(
        config=config,
        spawn_entries=entries,
        available_resources=frozenset({"res-a", "res-b"}),
    )
    assert not result.valid
    assert any("resource" in e.lower() for e in result.errors)
    print("validate-resource-ok")


def _validate_max_parallel() -> None:
    """Validate spawn exceeding max_parallel bound."""
    svc = SubplanService(decision_service=_mock_decision_service())
    config = SubplanConfig(
        execution_mode=ExecutionMode.PARALLEL,
        max_parallel=2,
    )
    entries = []
    for i in range(5):
        did = f"01HGZ6FE0AQDYTR4BXVQZ6D{i:02d}0"
        dec = _make_decision(did, sequence=i)
        entries.append(SpawnEntry(decision=dec, action_name=f"local/sub-{i}"))
    result = svc.validate_spawn(config=config, spawn_entries=entries)
    assert not result.valid
    assert any("max_parallel" in e.lower() for e in result.errors)
    print("validate-max-parallel-ok")


def _metadata_persistence() -> None:
    """Verify spawn metadata contains correct plan IDs."""
    svc = SubplanService(decision_service=_mock_decision_service())
    plan = _make_plan(
        parent_plan_id="01HGZ6FE0AQDYTR4BXVQZ6PP00",
        root_plan_id=_ROOT_ID,
    )
    config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL)
    dec = _make_decision(_DEC_ID1)
    entries = [SpawnEntry(decision=dec, action_name="local/sub-action")]
    result = svc.spawn(parent_plan=plan, config=config, spawn_entries=entries)
    meta = next(iter(result.metadata.values()))
    assert meta.root_plan_id == _ROOT_ID
    assert meta.execution_mode == "sequential"
    print("metadata-persistence-ok")


def _validation_guards() -> None:
    """Verify service rejects invalid construction and inputs."""
    # None decision_service
    try:
        SubplanService(decision_service=None)  # type: ignore[arg-type]
        raise AssertionError("Should have raised ValueError")
    except ValueError:
        pass

    # None parent_plan
    svc = SubplanService(decision_service=_mock_decision_service())
    try:
        svc.spawn(
            parent_plan=None,  # type: ignore[arg-type]
            config=SubplanConfig(),
            spawn_entries=[
                SpawnEntry(
                    decision=_make_decision(_DEC_ID1),
                    action_name="local/test",
                )
            ],
        )
        raise AssertionError("Should have raised ValueError")
    except ValueError:
        pass

    # Empty spawn_entries
    try:
        svc.spawn(
            parent_plan=_make_plan(),
            config=SubplanConfig(),
            spawn_entries=[],
        )
        raise AssertionError("Should have raised ValueError")
    except ValueError:
        pass

    print("validation-guards-ok")


def _build_entries() -> None:
    """Build spawn entries from decisions."""
    svc = SubplanService(decision_service=_mock_decision_service())
    decisions = [
        _make_decision(
            _DEC_ID1,
            resource_ids=["01HGZ6FE0AQDYTR4BXVQZ6RA00"],
            chosen_option="local/test-sub",
        ),
        _make_decision(
            _DEC_ID2,
            sequence=1,
            chosen_option="local/test-sub-2",
        ),
    ]
    entries = svc.build_spawn_entries(decisions)
    assert len(entries) == 2
    assert entries[0].action_name == "local/test-sub"
    assert len(entries[0].target_resources) == 1
    print("build-entries-ok")


_COMMANDS: dict[str, object] = {
    "spawn-single": _spawn_single,
    "spawn-multiple": _spawn_multiple,
    "validate-resource": _validate_resource_scope,
    "validate-max-parallel": _validate_max_parallel,
    "metadata-persistence": _metadata_persistence,
    "validation-guards": _validation_guards,
    "build-entries": _build_entries,
}

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)
    fn = _COMMANDS[sys.argv[1]]
    assert callable(fn)
    fn()