cleveragents-core/robot/helper_wf14_server_mode.py

"""Helper script for wf14_server_mode_integration.robot tests.

Each subcommand is a self-contained check that prints a sentinel on success.
Validates the server mode team collaboration workflow (Specification
Example 14) with mocked LLM providers and in-memory domain services.
"""

from __future__ import annotations

import logging
import os
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path

# Keep helper stderr focused on assertion failures and tracebacks.
logging.disable(logging.CRITICAL)

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

from cleveragents.application.services.config_service import (  # noqa: E402
    _REGISTRY,
    ConfigLevel,
    ConfigService,
)
from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.domain.models.core.action import (  # noqa: E402
    ActionArgument,
    ActionState,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    PlanPhase,
    ProcessingState,
    ProjectLink,
)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_config_service(tmpdir: str) -> ConfigService:
    """Create a ConfigService backed by a temporary directory."""
    config_dir = Path(tmpdir)
    config_path = config_dir / "config.toml"
    return ConfigService(config_dir=config_dir, config_path=config_path)


_WF14_CONFIG_ENV_OVERRIDES: tuple[str, ...] = (
    "CLEVERAGENTS_SERVER_URL",
    "CLEVERAGENTS_SERVER_TOKEN",
    "CLEVERAGENTS_SERVER_SYNC_AUTO",
    "CLEVERAGENTS_SERVER_SYNC_INTERVAL",
    "CLEVERAGENTS_SERVER_TLS_VERIFY",
    "CLEVERAGENTS_SERVER_NAMESPACE",
    "CLEVERAGENTS_CORE_NAMESPACE",
    "CLEVERAGENTS_NAMESPACE",
    "CLEVERAGENTS_CORE_AUTOMATION_PROFILE",
    "CLEVERAGENTS_AUTOMATION_PROFILE",
)


def _clear_config_env_overrides() -> None:
    """Remove env vars that can override config resolution in tests."""
    for env_name in _WF14_CONFIG_ENV_OVERRIDES:
        os.environ.pop(env_name, None)


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


def config_server_mode() -> None:
    """Set and verify server config values using ConfigService."""
    _clear_config_env_overrides()

    with tempfile.TemporaryDirectory() as tmpdir:
        svc = _make_config_service(tmpdir)

        # Set server mode config values
        svc.set_value("server.url", "https://agents.example.com")
        svc.set_value("server.token", "tok_wf14_test_abc123")
        svc.set_value("core.namespace", "team-alpha")

        # Read back and verify
        data = svc.read_config()
        assert data["server.url"] == "https://agents.example.com", (
            "Expected server.url='https://agents.example.com', "
            f"got {data.get('server.url')}"
        )
        assert data["server.token"] == "tok_wf14_test_abc123", (
            "Expected server.token='tok_wf14_test_abc123', "
            f"got {data.get('server.token')}"
        )
        assert data["core.namespace"] == "team-alpha", (
            f"Expected core.namespace='team-alpha', got {data.get('core.namespace')}"
        )

        # Verify resolution chain returns correct values
        resolved_url = svc.resolve("server.url")
        assert resolved_url.value == "https://agents.example.com", (
            "Expected resolve(server.url) to return configured URL, "
            f"got {resolved_url.value!r}"
        )
        assert resolved_url.source == ConfigLevel.GLOBAL, (
            "Expected resolve(server.url) source=global, "
            f"got {resolved_url.source.value}"
        )

        resolved_token = svc.resolve("server.token")
        assert resolved_token.value == "tok_wf14_test_abc123", (
            "Expected resolve(server.token) to return configured token, "
            f"got {resolved_token.value!r}"
        )
        assert resolved_token.source == ConfigLevel.GLOBAL, (
            "Expected resolve(server.token) source=global, "
            f"got {resolved_token.source.value}"
        )

        resolved_ns = svc.resolve("core.namespace")
        assert resolved_ns.value == "team-alpha", (
            "Expected resolve(core.namespace) to return configured namespace, "
            f"got {resolved_ns.value!r}"
        )
        assert resolved_ns.source == ConfigLevel.GLOBAL, (
            "Expected resolve(core.namespace) source=global, "
            f"got {resolved_ns.source.value}"
        )

    print("wf14-config-server-mode-ok")


def diagnostics() -> None:
    """Verify config registry contains required server-mode keys and defaults.

    Checks that all server-mode configuration keys are registered and that
    their default values match the specification.  Server connectivity and
    database checks are not exercised here because no real server or
    database is available in the integration test environment.
    """
    _clear_config_env_overrides()

    # Verify config registry has required server keys
    required_keys = [
        "server.url",
        "server.token",
        "server.sync.auto",
        "server.sync.interval",
        "server.tls-verify",
        "server.namespace",
        "core.namespace",
        "core.automation-profile",
    ]
    for key in required_keys:
        assert key in _REGISTRY, f"Missing config key: {key}"

    # Verify config service can be instantiated with temp directory
    with tempfile.TemporaryDirectory() as tmpdir:
        svc = _make_config_service(tmpdir)

        # Verify basic config reads succeed (empty config initially)
        data = svc.read_config()
        assert len(data) == 0, (
            "Expected empty initial config in temporary directory, "
            f"got keys: {sorted(data.keys())}"
        )

        # Verify default values are accessible via resolution
        resolved = svc.resolve("server.sync.auto")
        assert resolved.value is True, f"Expected True, got {resolved.value}"
        assert resolved.source == ConfigLevel.DEFAULT, (
            f"Expected server.sync.auto source=default, got {resolved.source.value}"
        )

        resolved_interval = svc.resolve("server.sync.interval")
        assert resolved_interval.value == 300, (
            f"Expected 300, got {resolved_interval.value}"
        )
        assert resolved_interval.source == ConfigLevel.DEFAULT, (
            "Expected server.sync.interval source=default, "
            f"got {resolved_interval.source.value}"
        )

        resolved_tls = svc.resolve("server.tls-verify")
        assert resolved_tls.value is True, f"Expected True, got {resolved_tls.value}"
        assert resolved_tls.source == ConfigLevel.DEFAULT, (
            "Expected server.tls-verify source=default, "
            f"got {resolved_tls.source.value}"
        )

        # Verify core.namespace default is 'local'
        resolved_ns = svc.resolve("core.namespace")
        assert resolved_ns.value == "local", (
            f"Expected 'local', got {resolved_ns.value}"
        )
        assert resolved_ns.source == ConfigLevel.DEFAULT, (
            f"Expected core.namespace source=default, got {resolved_ns.source.value}"
        )

        # Verify automation-profile default is 'supervised'
        resolved_profile = svc.resolve("core.automation-profile")
        assert resolved_profile.value == "supervised", (
            f"Expected 'supervised', got {resolved_profile.value}"
        )
        assert resolved_profile.source == ConfigLevel.DEFAULT, (
            "Expected core.automation-profile source=default, "
            f"got {resolved_profile.source.value}"
        )

        # Server connectivity cannot be tested (no real server) -- verify
        # that the server.url default is None (graceful handling)
        resolved_url = svc.resolve("server.url")
        assert resolved_url.value is None, (
            f"Expected None default for server.url, got {resolved_url.value}"
        )
        assert resolved_url.source == ConfigLevel.DEFAULT, (
            f"Expected server.url source=default, got {resolved_url.source.value}"
        )

    print("wf14-diagnostics-ok")


def action_namespace() -> None:
    """Create and list namespaced actions using PlanLifecycleService."""
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)
    definition_of_done = "Service is running and health checks pass"

    # Create an action with a namespace prefix
    action = lifecycle.create_action(
        name="team-alpha/deploy-service",
        description="Deploy a microservice to staging",
        definition_of_done=definition_of_done,
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
        tags=["deploy", "team-alpha"],
    )

    assert str(action.namespaced_name) == "team-alpha/deploy-service", (
        f"Expected 'team-alpha/deploy-service', got '{action.namespaced_name}'"
    )
    assert action.state == ActionState.AVAILABLE, "Expected action state AVAILABLE"
    assert action.reusable is True, "Expected action.reusable to default to True"
    assert action.read_only is False, "Expected action.read_only to default to False"
    assert action.automation_profile == "supervised", (
        "Expected action automation_profile='supervised', "
        f"got {action.automation_profile!r}"
    )
    assert action.definition_of_done == definition_of_done, (
        "Expected definition_of_done to match the configured value"
    )
    assert action.tags == ["deploy", "team-alpha"], (
        f"Expected action tags ['deploy', 'team-alpha'], got {action.tags!r}"
    )

    # List actions and verify the namespaced action appears
    actions = lifecycle.list_actions()
    assert len(actions) == 1, f"Expected 1 action, got {len(actions)}"
    action_names = [str(a.namespaced_name) for a in actions]
    assert "team-alpha/deploy-service" in action_names, (
        f"'team-alpha/deploy-service' not in {action_names}"
    )

    print("wf14-action-namespace-ok")


def action_actor_refs() -> None:
    """Create an action with namespaced actor references and verify fields.

    Validates that actions can reference actors from different namespaces
    (strategy, execution, review) and that those references are preserved
    when the action is retrieved.
    """
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)

    # Create an action that references namespaced actors
    action = lifecycle.create_action(
        name="team-alpha/analyze-code",
        description="Analyze codebase for improvements",
        definition_of_done="Analysis report generated",
        strategy_actor="openai/gpt-4",
        execution_actor="anthropic/claude-sonnet",
        review_actor="openai/gpt-4",
        automation_profile="supervised",
    )

    assert action.strategy_actor == "openai/gpt-4", (
        f"Expected strategy_actor 'openai/gpt-4', got {action.strategy_actor!r}"
    )
    assert action.execution_actor == "anthropic/claude-sonnet", (
        "Expected execution_actor 'anthropic/claude-sonnet', "
        f"got {action.execution_actor!r}"
    )
    assert action.review_actor == "openai/gpt-4", (
        f"Expected review_actor 'openai/gpt-4', got {action.review_actor!r}"
    )
    assert action.automation_profile == "supervised", (
        "Expected action automation_profile='supervised', "
        f"got {action.automation_profile!r}"
    )

    # Verify the action is accessible
    fetched = lifecycle.get_action("team-alpha/analyze-code")
    assert str(fetched.namespaced_name) == "team-alpha/analyze-code", (
        "Expected fetched action name 'team-alpha/analyze-code', "
        f"got {fetched.namespaced_name!s}"
    )
    assert fetched.strategy_actor == "openai/gpt-4", (
        "Expected fetched strategy_actor 'openai/gpt-4', "
        f"got {fetched.strategy_actor!r}"
    )
    assert fetched.execution_actor == "anthropic/claude-sonnet", (
        "Expected fetched execution_actor 'anthropic/claude-sonnet', "
        f"got {fetched.execution_actor!r}"
    )
    assert fetched.review_actor == "openai/gpt-4", (
        f"Expected fetched review_actor 'openai/gpt-4', got {fetched.review_actor!r}"
    )

    print("wf14-action-actor-refs-ok")


def shared_actions() -> None:
    """List actions from a specific namespace."""
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)

    # Create actions in different namespaces
    lifecycle.create_action(
        name="team-alpha/shared-lint",
        description="Lint code with team standards",
        definition_of_done="All lint checks pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
    )
    lifecycle.create_action(
        name="team-beta/shared-test",
        description="Run integration test suite",
        definition_of_done="All tests pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
    )
    lifecycle.create_action(
        name="team-alpha/shared-review",
        description="Code review with team conventions",
        definition_of_done="Review complete with no blockers",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
    )

    # Verify total unfiltered count
    all_actions = lifecycle.list_actions()
    assert len(all_actions) == 3, f"Expected 3 total actions, got {len(all_actions)}"
    assert all(a.state == ActionState.AVAILABLE for a in all_actions), (
        "Expected all actions to be AVAILABLE"
    )

    # List only team-alpha namespace
    alpha_actions = lifecycle.list_actions(namespace="team-alpha")
    assert len(alpha_actions) == 2, (
        f"Expected 2 team-alpha actions, got {len(alpha_actions)}"
    )
    alpha_names = [str(a.namespaced_name) for a in alpha_actions]
    assert "team-alpha/shared-lint" in alpha_names, (
        f"'team-alpha/shared-lint' not in {alpha_names}"
    )
    assert "team-alpha/shared-review" in alpha_names, (
        f"'team-alpha/shared-review' not in {alpha_names}"
    )
    # Ensure only team-alpha namespace values are returned.
    assert all(n.startswith("team-alpha/") for n in alpha_names), (
        f"Expected only team-alpha names, got {alpha_names}"
    )

    # List team-beta namespace
    beta_actions = lifecycle.list_actions(namespace="team-beta")
    beta_names = [str(a.namespaced_name) for a in beta_actions]
    assert "team-beta/shared-test" in beta_names, (
        f"'team-beta/shared-test' not in {beta_names}"
    )
    assert len(beta_actions) == 1, f"Expected 1 beta action, got {len(beta_actions)}"

    print("wf14-shared-actions-ok")


def use_shared_action() -> None:
    """Use a shared action to create a plan and verify plan attributes.

    Covers Specification Example 14, Step 3: a developer uses a shared
    action published by another team member.  Verifies that the resulting
    plan inherits the action's namespace and enters the expected initial
    phase and processing state.
    """
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)

    # Publish an action (simulating another team member)
    target_module_arg = ActionArgument.parse(
        "target_module:string:required:Module path to generate tests"
    )
    published = lifecycle.create_action(
        name="team-alpha/generate-tests",
        description="Generate tests for a target module",
        definition_of_done="All target modules have test coverage above 80%",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
        arguments=[target_module_arg],
        tags=["wf14", "shared"],
    )
    assert published.automation_profile == "supervised", (
        "Expected published action automation_profile='supervised', "
        f"got {published.automation_profile!r}"
    )

    # Use the shared action (simulating a second developer)
    plan = lifecycle.use_action(
        action_name="team-alpha/generate-tests",
        project_links=[ProjectLink(project_name="local/payment-service")],
        arguments={"target_module": "src/payments"},
    )

    # Verify plan attributes
    assert plan.namespaced_name.namespace == "team-alpha", (
        f"Expected namespace 'team-alpha', got '{plan.namespaced_name.namespace}'"
    )
    assert plan.phase == PlanPhase.STRATEGIZE, (
        f"Expected phase STRATEGIZE, got {plan.phase}"
    )
    assert plan.processing_state == ProcessingState.COMPLETE, (
        f"Expected state COMPLETE, got {plan.processing_state}"
    )
    assert plan.action_name == "team-alpha/generate-tests", (
        f"Expected action_name 'team-alpha/generate-tests', got '{plan.action_name}'"
    )
    assert plan.strategy_actor == "openai/gpt-4", (
        f"Expected strategy_actor 'openai/gpt-4', got {plan.strategy_actor!r}"
    )
    assert plan.execution_actor == "openai/gpt-4", (
        f"Expected execution_actor 'openai/gpt-4', got {plan.execution_actor!r}"
    )
    assert plan.arguments == {"target_module": "src/payments"}, (
        "Expected plan arguments {'target_module': 'src/payments'}, "
        f"got {plan.arguments!r}"
    )
    assert plan.arguments_order == ["target_module"], (
        f"Expected arguments_order ['target_module'], got {plan.arguments_order!r}"
    )
    assert len(plan.project_links) == 1, (
        f"Expected 1 project link, got {len(plan.project_links)}"
    )
    assert plan.project_links[0].project_name == "local/payment-service", (
        "Expected project link to local/payment-service, "
        f"got {plan.project_links[0].project_name!r}"
    )
    assert plan.definition_of_done == published.definition_of_done, (
        "Expected plan definition_of_done to match published action"
    )
    assert plan.reusable is True, "Expected plan.reusable=True from action default"
    assert plan.read_only is False, "Expected plan.read_only=False from action default"
    assert plan.tags == ["wf14", "shared"], (
        f"Expected plan tags ['wf14', 'shared'], got {plan.tags!r}"
    )

    print("wf14-use-shared-action-ok")


def plan_namespace() -> None:
    """Create plans and verify namespace-filtered listing.

    Covers Specification Example 14, Step 4: monitoring plans across the
    team by listing plans filtered by namespace.  Creates actions in two
    namespaces, uses them to instantiate plans, and verifies that namespace
    filtering returns only the expected plans.
    """
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)

    # Create and use actions in team-alpha namespace
    lifecycle.create_action(
        name="team-alpha/monitor-lint",
        description="Lint code with team standards",
        definition_of_done="All lint checks pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
    )
    lifecycle.create_action(
        name="team-beta/monitor-test",
        description="Run integration test suite",
        definition_of_done="All tests pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        automation_profile="supervised",
    )

    alpha_plan = lifecycle.use_action(action_name="team-alpha/monitor-lint")
    beta_plan = lifecycle.use_action(action_name="team-beta/monitor-test")

    # Both plans auto-completed strategize (decompose_task=0.0 in supervised profile)
    # Move beta to Execute — already at STRATEGIZE/COMPLETE, skip to execute_plan
    beta_plan = lifecycle.execute_plan(beta_plan.identity.plan_id)

    # List all plans
    all_plans = lifecycle.list_plans()
    assert len(all_plans) == 2, f"Expected 2 plans total, got {len(all_plans)}"

    # List plans filtered to team-alpha namespace
    alpha_plans = lifecycle.list_plans(namespace="team-alpha")
    assert len(alpha_plans) == 1, f"Expected 1 team-alpha plan, got {len(alpha_plans)}"
    assert alpha_plans[0].identity.plan_id == alpha_plan.identity.plan_id, (
        "team-alpha plan ID mismatch"
    )
    assert alpha_plans[0].phase == PlanPhase.STRATEGIZE, (
        f"Expected team-alpha phase STRATEGIZE, got {alpha_plans[0].phase}"
    )
    assert alpha_plans[0].processing_state == ProcessingState.COMPLETE, (
        "Expected team-alpha processing state COMPLETE, "
        f"got {alpha_plans[0].processing_state}"
    )
    assert alpha_plans[0].action_name == "team-alpha/monitor-lint", (
        "Expected team-alpha action_name 'team-alpha/monitor-lint', "
        f"got {alpha_plans[0].action_name!r}"
    )

    # List plans filtered to team-beta namespace
    beta_plans = lifecycle.list_plans(namespace="team-beta")
    assert len(beta_plans) == 1, f"Expected 1 team-beta plan, got {len(beta_plans)}"
    assert beta_plans[0].identity.plan_id == beta_plan.identity.plan_id, (
        "team-beta plan ID mismatch"
    )
    assert beta_plans[0].phase == PlanPhase.EXECUTE, (
        f"Expected team-beta phase EXECUTE, got {beta_plans[0].phase}"
    )
    assert beta_plans[0].processing_state == ProcessingState.QUEUED, (
        "Expected team-beta processing state QUEUED after execute transition, "
        f"got {beta_plans[0].processing_state}"
    )
    assert beta_plans[0].action_name == "team-beta/monitor-test", (
        "Expected team-beta action_name 'team-beta/monitor-test', "
        f"got {beta_plans[0].action_name!r}"
    )

    # Verify phase filtering within namespaces.
    alpha_strategize = lifecycle.list_plans(
        namespace="team-alpha", phase=PlanPhase.STRATEGIZE
    )
    assert len(alpha_strategize) == 1, (
        f"Expected 1 team-alpha strategize plan, got {len(alpha_strategize)}"
    )
    assert alpha_strategize[0].identity.plan_id == alpha_plan.identity.plan_id, (
        "team-alpha strategize phase filter returned wrong plan"
    )

    beta_execute = lifecycle.list_plans(namespace="team-beta", phase=PlanPhase.EXECUTE)
    assert len(beta_execute) == 1, (
        f"Expected 1 team-beta execute plan, got {len(beta_execute)}"
    )
    assert beta_execute[0].identity.plan_id == beta_plan.identity.plan_id, (
        "team-beta execute phase filter returned wrong plan"
    )

    # Verify cross-namespace isolation
    gamma_plans = lifecycle.list_plans(namespace="team-gamma")
    assert len(gamma_plans) == 0, f"Expected 0 team-gamma plans, got {len(gamma_plans)}"

    print("wf14-plan-namespace-ok")


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "config-server-mode": config_server_mode,
    "diagnostics": diagnostics,
    "action-namespace": action_namespace,
    "action-actor-refs": action_actor_refs,
    "shared-actions": shared_actions,
    "use-shared-action": use_shared_action,
    "plan-namespace": plan_namespace,
}

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