cleveragents-core/robot/helper_plan_execute_runtime.py

"""Helper script for plan execute runtime Robot Framework tests.

Dispatches to individual test functions via a COMMANDS dict.
"""

from __future__ import annotations

import sys

from ulid import ULID

from cleveragents.application.services.plan_execution_context import (
    PlanExecutionContext,
    RuntimeExecuteActor,
    RuntimeExecuteResult,
)
from cleveragents.application.services.plan_executor import (
    PlanExecutor,
    StrategyDecision,
)
from cleveragents.core.exceptions import ValidationError
from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner


def _make_runner() -> ToolRunner:
    return ToolRunner(registry=ToolRegistry())


def _make_decisions(count: int) -> list[StrategyDecision]:
    root_id = str(ULID())
    return [
        StrategyDecision(
            decision_id=root_id if i == 0 else str(ULID()),
            step_text=f"Step {i + 1}",
            sequence=i,
            parent_id=root_id if i > 0 else None,
        )
        for i in range(count)
    ]


def _context_creation() -> None:
    """Test PlanExecutionContext creation and defaults."""
    plan_id = str(ULID())
    ctx = PlanExecutionContext(plan_id=plan_id)
    assert ctx.plan_id == plan_id
    assert ctx.decision_root_id is None
    assert ctx.sandbox_root is None
    assert ctx.automation_profile is None
    assert ctx.project_resources == {}
    assert ctx.resource_bindings == {}
    assert isinstance(ctx.changeset_store, InMemoryChangeSetStore)

    # With all fields
    store = InMemoryChangeSetStore()
    ctx2 = PlanExecutionContext(
        plan_id=plan_id,
        decision_root_id="root-123",
        sandbox_root="/tmp/sb",
        automation_profile="trusted",
        project_resources={"r": "v"},
        changeset_store=store,
    )
    assert ctx2.decision_root_id == "root-123"
    assert ctx2.sandbox_root == "/tmp/sb"
    assert ctx2.automation_profile == "trusted"
    assert ctx2.changeset_store is store

    # Empty plan_id should fail
    try:
        PlanExecutionContext(plan_id="")
        raise AssertionError("Expected ValidationError")
    except ValidationError:
        pass

    print("context-creation-ok")


def _changeset_ops() -> None:
    """Test changeset start, record, get, summarize."""
    plan_id = str(ULID())
    store = InMemoryChangeSetStore()
    ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store)

    cs_id = ctx.start_changeset()
    assert len(cs_id) == 26  # ULID length
    assert cs_id in ctx.active_changeset_ids

    entry = ChangeEntry(
        plan_id=plan_id,
        resource_id=str(ULID()),
        tool_name="test/tool",
        operation=ChangeOperation.MODIFY,
        path="src/test.py",
    )
    ctx.record_change(entry)

    cs = ctx.get_changeset(cs_id)
    assert cs is not None
    assert len(cs.entries) == 1
    assert cs.entries[0].entry_id == entry.entry_id

    # Unknown ID
    assert ctx.get_changeset("nonexistent") is None

    # Summarize
    summary = ctx.summarize()
    assert summary["plan_id"] == plan_id
    assert summary["active_changeset_count"] == 1
    assert len(summary["changeset_summaries"]) == 1

    print("changeset-ops-ok")


def _runtime_actor() -> None:
    """Test RuntimeExecuteActor execution."""
    plan_id = str(ULID())
    store = InMemoryChangeSetStore()
    ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store)
    runner = _make_runner()
    actor = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx)

    decisions = _make_decisions(2)
    result = actor.execute(decisions=decisions)

    assert isinstance(result, RuntimeExecuteResult)
    assert len(result.changeset_id) == 26
    assert len(result.decision_ids_processed) == 2
    assert result.execution_duration_ms >= 0.0

    # With stream callback
    events: list[tuple[str, dict]] = []
    actor2 = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx)
    actor2.execute(
        decisions=_make_decisions(1),
        stream_callback=lambda t, d: events.append((t, d)),
    )
    types = [e[0] for e in events]
    assert "runtime_execute_started" in types
    assert "runtime_execute_complete" in types

    print("runtime-actor-ok")


def _executor_runtime() -> None:
    """Test PlanExecutor with execution context (runtime mode)."""
    plan_id = str(ULID())
    store = InMemoryChangeSetStore()
    ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store)

    from unittest.mock import MagicMock

    lifecycle = MagicMock()

    executor = PlanExecutor(
        lifecycle_service=lifecycle,
        tool_runner=_make_runner(),
        execution_context=ctx,
    )
    assert executor.has_runtime is True
    assert executor.changeset_store is store
    assert executor.execution_context is ctx

    print("executor-runtime-ok")


def _executor_stub() -> None:
    """Test PlanExecutor without execution context (stub mode)."""
    from unittest.mock import MagicMock

    lifecycle = MagicMock()

    executor = PlanExecutor(lifecycle_service=lifecycle)
    assert executor.has_runtime is False
    assert executor.changeset_store is None
    assert executor.execution_context is None

    print("executor-stub-ok")


COMMANDS: dict[str, object] = {
    "context-creation": _context_creation,
    "changeset-ops": _changeset_ops,
    "runtime-actor": _runtime_actor,
    "executor-runtime": _executor_runtime,
    "executor-stub": _executor_stub,
}


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit("Expected command argument")
    command = sys.argv[1]
    if command not in COMMANDS:
        raise SystemExit(f"Unknown command: {command}")
    func = COMMANDS[command]
    if callable(func):
        func()


if __name__ == "__main__":
    main()