cleveragents-core/robot/helper_context_strategy_registry.py

#!/usr/bin/env python3
"""Robot Framework helper for context strategy registry smoke tests."""

from __future__ import annotations

import sys
from collections.abc import Callable
from pathlib import Path

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

from cleveragents.application.services.strategy_registry import (  # noqa: E402
    StrategyRegistry,
)
from cleveragents.domain.models.acms.crp import ContextRequest  # noqa: E402
from cleveragents.domain.models.acms.strategy import (  # noqa: E402
    BackendSet,
    ContextStrategy,
    StrategyConfig,
)
from cleveragents.domain.models.acms.strategy_stubs import (  # noqa: E402
    BUILTIN_STRATEGY_CLASSES,
    DEFAULT_ENABLED_STRATEGIES,
)
from cleveragents.domain.models.acms.stubs import (  # noqa: E402
    InMemoryGraphBackend,
    InMemoryTextBackend,
    InMemoryVectorBackend,
)

# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------


def _cmd_register_builtins() -> int:
    """Register all built-in strategies and verify count."""
    registry = StrategyRegistry()
    for cls in BUILTIN_STRATEGY_CLASSES:
        inst = cls()
        enabled = inst.name in DEFAULT_ENABLED_STRATEGIES
        registry.register(inst, config=StrategyConfig(enabled=enabled), is_builtin=True)
    registry.set_enabled(list(DEFAULT_ENABLED_STRATEGIES))

    if len(registry) != 6:
        print(f"strategy-fail: expected 6 strategies, got {len(registry)}")
        return 1
    print(f"strategy-ok: registered {len(registry)} strategies")

    enabled = registry.list_enabled()
    if len(enabled) != 3:
        print(f"strategy-fail: expected 3 enabled, got {len(enabled)}")
        return 1
    print(f"strategy-enabled: {enabled}")
    return 0


def _cmd_protocol_check() -> int:
    """Verify all built-in strategies satisfy ContextStrategy protocol."""
    for cls in BUILTIN_STRATEGY_CLASSES:
        inst = cls()
        if not isinstance(inst, ContextStrategy):
            print(f"strategy-fail: {cls.__name__} doesn't satisfy protocol")
            return 1
        if not inst.name:
            print(f"strategy-fail: {cls.__name__} has empty name")
            return 1
        if inst.capabilities.quality_score <= 0:
            print(f"strategy-fail: {cls.__name__} has invalid quality score")
            return 1
        print(f"strategy-ok: {inst.name} (quality={inst.capabilities.quality_score})")
    return 0


def _cmd_can_handle() -> int:
    """Test can_handle with various backend configurations."""
    from cleveragents.domain.models.acms.strategy_stubs import (
        ARCEStrategy,
        SimpleKeywordStrategy,
    )

    request = ContextRequest(query="test")

    # simple-keyword with text backend
    sk = SimpleKeywordStrategy()
    bs_text = BackendSet(text=InMemoryTextBackend())
    score = sk.can_handle(request, bs_text)
    if abs(score - 0.3) > 1e-6:
        print(f"strategy-fail: simple-keyword expected 0.3, got {score}")
        return 1
    print(f"strategy-ok: simple-keyword can_handle with text = {score}")

    # simple-keyword without backend
    bs_none = BackendSet()
    score = sk.can_handle(request, bs_none)
    if score != 0.0:
        print(
            f"strategy-fail: simple-keyword expected 0.0 without backend, got {score}"
        )
        return 1
    print(f"strategy-ok: simple-keyword can_handle without backend = {score}")

    # arce with all backends
    arce = ARCEStrategy()
    bs_all = BackendSet(
        text=InMemoryTextBackend(),
        vector=InMemoryVectorBackend(),
        graph=InMemoryGraphBackend(),
    )
    score = arce.can_handle(request, bs_all)
    if abs(score - 0.95) > 1e-6:
        print(f"strategy-fail: arce expected 0.95, got {score}")
        return 1
    print(f"strategy-ok: arce can_handle with all = {score}")

    return 0


def _cmd_validate() -> int:
    """Validate registry with all built-in strategies."""
    registry = StrategyRegistry()
    for cls in BUILTIN_STRATEGY_CLASSES:
        registry.register(cls(), is_builtin=True)

    warnings = registry.validate_registry()
    if warnings:
        print(f"strategy-fail: unexpected warnings: {warnings}")
        return 1
    print("strategy-ok: registry validation passed with no warnings")
    return 0


_COMMANDS: dict[str, Callable[[], int]] = {
    "register-builtins": _cmd_register_builtins,
    "protocol-check": _cmd_protocol_check,
    "can-handle": _cmd_can_handle,
    "validate": _cmd_validate,
}


def main() -> int:
    """Run the specified command."""
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
        return 2
    return _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    sys.exit(main())