temp/robot/helper_tdd_subplan_spawn_orchestration.py

"""Helper script for tdd_subplan_spawn_orchestration.robot smoke tests.

Each subcommand exercises SubplanService.spawn() to reproduce bug #823.
The helper reports the **real** outcome: it exits 0 and prints the sentinel
when the expected behaviour is observed (bug fixed), and exits 1 when the
bug is still present.  The ``tdd_expected_fail_listener`` on the Robot side
handles pass/fail inversion while the bug remains open.
"""

from __future__ import annotations

import sys
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn
from unittest.mock import MagicMock

# Ensure local source tree is importable.
_ROOT: Path = Path(__file__).resolve().parents[1]
_SRC: str = str(_ROOT / "src")
_ROBOT: str = str(_ROOT / "robot")
for _p in (_SRC, _ROBOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from cleveragents.application.services.subplan_service import (  # noqa: E402
    SpawnEntry,
    SpawnResult,
    SubplanService,
)
from cleveragents.domain.models.core.decision import (  # noqa: E402
    ContextSnapshot,
    Decision,
    DecisionType,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    ExecutionMode,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    ProcessingState,
    SubplanConfig,
)

_PARENT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
_ROOT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
_DEC_ID1: str = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
_DEC_ID2: str = "01HGZ6FE0AQDYTR4BXVQZ6DB00"


def _fail(msg: str) -> NoReturn:
    """Print error message to stderr and exit with code 1."""
    print(msg, file=sys.stderr)
    sys.exit(1)


def _mock_decision_service() -> MagicMock:
    """Create a mock DecisionService."""
    svc: MagicMock = MagicMock()
    svc.list_by_type = MagicMock(return_value=[])
    return svc


def _make_parent_plan() -> Plan:
    """Create a parent plan suitable for subplan spawning."""
    return Plan(
        identity=PlanIdentity(
            plan_id=_PARENT_PLAN_ID,
            root_plan_id=_ROOT_PLAN_ID,
        ),
        namespaced_name=NamespacedName(namespace="local", name="parent-plan"),
        description="Parent plan for subplan spawn orchestration test",
        action_name="local/parent-action",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.PROCESSING,
        subplan_config=SubplanConfig(
            execution_mode=ExecutionMode.SEQUENTIAL,
        ),
    )


def _make_spawn_entries() -> list[SpawnEntry]:
    """Create spawn entries for testing."""
    dec1: Decision = Decision(
        decision_id=_DEC_ID1,
        plan_id=_PARENT_PLAN_ID,
        decision_type=DecisionType.SUBPLAN_SPAWN,
        sequence_number=0,
        question="Spawn child plan for module A?",
        chosen_option="local/sub-action-a",
        context_snapshot=ContextSnapshot(),
    )
    dec2: Decision = Decision(
        decision_id=_DEC_ID2,
        plan_id=_PARENT_PLAN_ID,
        decision_type=DecisionType.SUBPLAN_SPAWN,
        sequence_number=1,
        question="Spawn child plan for module B?",
        chosen_option="local/sub-action-b",
        context_snapshot=ContextSnapshot(),
    )
    return [
        SpawnEntry(decision=dec1, action_name="local/sub-action-a"),
        SpawnEntry(decision=dec2, action_name="local/sub-action-b"),
    ]


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def _spawn_child_plans() -> None:
    """Verify spawn() returns actual child Plan domain objects.

    Bug #823: SpawnResult only has spawned_statuses (SubplanStatus) and
    metadata (SpawnMetadata).  No child_plans attribute exists.
    """
    svc: SubplanService = SubplanService(
        decision_service=_mock_decision_service(),
    )
    parent: Plan = _make_parent_plan()
    config: SubplanConfig = SubplanConfig(
        execution_mode=ExecutionMode.SEQUENTIAL,
    )
    entries: list[SpawnEntry] = _make_spawn_entries()

    result: SpawnResult = svc.spawn(
        parent_plan=parent,
        config=config,
        spawn_entries=entries,
    )

    # Assert that SpawnResult contains child Plan objects
    child_plans: object = getattr(result, "child_plans", None)
    if child_plans is None:
        _fail(
            "SpawnResult does not contain child_plans attribute — "
            "spawn() only creates metadata without actual child Plan "
            "objects (bug #823)"
        )
    if not isinstance(child_plans, list):
        _fail("SpawnResult.child_plans should be a list of Plan objects")
    if len(child_plans) != len(entries):
        _fail(f"Expected {len(entries)} child plans, got {len(child_plans)}")
    # Verify each child has correct parent_plan_id
    for child in child_plans:
        if not isinstance(child, Plan):
            _fail(f"Expected Plan instance, got {type(child).__name__}")
        if child.identity.parent_plan_id != _PARENT_PLAN_ID:
            _fail(
                f"Child plan parent_plan_id={child.identity.parent_plan_id!r}, "
                f"expected {_PARENT_PLAN_ID!r}"
            )
    print("tdd-spawn-child-plans-ok")


def _spawn_lifecycle() -> None:
    """Verify child plans enter the strategize phase after spawn.

    Bug #823: No child Plan objects are created, so lifecycle is not
    triggered.
    """
    svc: SubplanService = SubplanService(
        decision_service=_mock_decision_service(),
    )
    parent: Plan = _make_parent_plan()
    config: SubplanConfig = SubplanConfig(
        execution_mode=ExecutionMode.SEQUENTIAL,
    )
    entries: list[SpawnEntry] = _make_spawn_entries()

    result: SpawnResult = svc.spawn(
        parent_plan=parent,
        config=config,
        spawn_entries=entries,
    )

    # Assert that child plans exist and are in strategize phase
    child_plans: object = getattr(result, "child_plans", None)
    if child_plans is None:
        _fail(
            "No child_plans on SpawnResult — spawn() does not create "
            "child Plan objects or trigger lifecycle (bug #823)"
        )
    if not isinstance(child_plans, list):
        _fail("child_plans is not a list")
    for child in child_plans:
        if not isinstance(child, Plan):
            _fail(f"Expected Plan, got {type(child).__name__}")
        if child.phase != PlanPhase.STRATEGIZE:
            _fail(f"Child plan phase={child.phase!r}, expected STRATEGIZE")
        if child.processing_state != ProcessingState.QUEUED:
            _fail(f"Child plan state={child.processing_state!r}, expected QUEUED")
    print("tdd-spawn-lifecycle-ok")


def _spawn_parent_tracking() -> None:
    """Verify parent plan tracks spawned child plan statuses.

    Bug #823: spawn() creates SubplanStatus records in SpawnResult but
    does not attach them to the parent plan's subplan_statuses list.
    """
    svc: SubplanService = SubplanService(
        decision_service=_mock_decision_service(),
    )
    parent: Plan = _make_parent_plan()
    config: SubplanConfig = SubplanConfig(
        execution_mode=ExecutionMode.SEQUENTIAL,
    )
    entries: list[SpawnEntry] = _make_spawn_entries()

    result: SpawnResult = svc.spawn(
        parent_plan=parent,
        config=config,
        spawn_entries=entries,
    )

    # After spawn(), parent plan's subplan_statuses should be populated
    if len(parent.subplan_statuses) != len(entries):
        _fail(
            f"Parent plan has {len(parent.subplan_statuses)} subplan_statuses "
            f"but {len(entries)} were spawned — spawn() does not attach "
            f"statuses to the parent plan (bug #823). "
            f"Statuses only in SpawnResult.spawned_statuses "
            f"({result.total_spawned} entries)."
        )
    print("tdd-spawn-parent-tracking-ok")


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "spawn-child-plans": _spawn_child_plans,
    "spawn-lifecycle": _spawn_lifecycle,
    "spawn-parent-tracking": _spawn_parent_tracking,
}

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)
    cmd: Callable[[], None] = _COMMANDS[sys.argv[1]]
    cmd()