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


def _cmd_pipeline_integration() -> int:
    """Integration test: verify all 6 spec strategies are registered in ACMSPipeline.

    Verifies that the 6 spec-required strategies are registered with ACMSPipeline
    and that the pipeline can be used with the spec strategy names.
    """
    from cleveragents.application.services.acms_service import ACMSPipeline

    pipeline = ACMSPipeline()

    # Verify all 6 spec-required strategies are registered
    expected_strategies = {
        "simple-keyword",
        "semantic-embedding",
        "breadth-depth-navigator",
        "arce",
        "temporal-archaeology",
        "plan-decision-context",
    }
    registered = set(pipeline._strategies.keys())
    missing = expected_strategies - registered
    if missing:
        print(f"strategy-fail: missing strategies: {missing}")
        return 1

    print("strategy-ok: all 6 spec strategies registered in ACMSPipeline")
    print(f"strategy-ok: total strategies: {len(registered)}")
    return 0


def _cmd_real_retrieval() -> int:
    """Test that each strategy returns non-empty fragments with populated backends."""
    from datetime import UTC, datetime, timedelta

    from cleveragents.domain.models.acms.backends import (
        GraphResult,
        TextResult,
        VectorResult,
    )
    from cleveragents.domain.models.acms.crp import ContextRequest
    from cleveragents.domain.models.acms.strategy import BackendSet, PlanContext
    from cleveragents.domain.models.acms.strategy_stubs import BUILTIN_STRATEGY_CLASSES
    from cleveragents.domain.models.acms.temporal import TemporalMetadata, TemporalNode
    from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend

    class _TextBackend:
        def search(self, query, *, scope, max_results=20):
            return [
                TextResult(
                    uko_uri="uko:class/Auth",
                    content=f"Auth for {query}",
                    score=0.9,
                )
            ]

    class _VectorBackend:
        def similarity_search(self, embedding, *, scope, top_k=20):
            return [
                VectorResult(
                    uko_uri="uko:class/Auth",
                    content="Auth semantic",
                    score=0.85,
                )
            ]

    class _GraphBackend:
        def sparql_query(self, query, *, scope):
            return GraphResult(triples=[("uko:class/Auth", "rdf:type", "uko:Class")])

        def get_triples(self, subject):
            return GraphResult(triples=[(subject, "rdf:type", "uko:Class")])

        def traverse(self, start, *, depth=2):
            return GraphResult(
                triples=[
                    (start, "rdf:type", "uko:Class"),
                    (start, "uko:hasMethod", "uko:method/auth"),
                ]
            )

    # Create temporal backend with data
    temporal_backend = InMemoryTemporalBackend()
    now = datetime.now(tz=UTC)
    node = TemporalNode(
        node_uri="uko:plan/test_v1",
        source_resource="res://test",
        source_path="src/test.py",
        temporal=TemporalMetadata(
            valid_from=now - timedelta(hours=1),
            is_current=True,
        ),
    )
    temporal_backend.store_node(node)

    request = ContextRequest(query="authentication", focus=["uko:class/Auth"])
    plan_context = PlanContext()

    backend_map = {
        "simple-keyword": BackendSet(text=_TextBackend()),
        "semantic-embedding": BackendSet(vector=_VectorBackend()),
        "breadth-depth-navigator": BackendSet(graph=_GraphBackend()),
        "arce": BackendSet(
            text=_TextBackend(),
            vector=_VectorBackend(),
            graph=_GraphBackend(),
        ),
        "temporal-archaeology": BackendSet(
            graph=_GraphBackend(),
            temporal=temporal_backend,
        ),
        "plan-decision-context": BackendSet(temporal=temporal_backend),
    }

    all_ok = True
    for cls in BUILTIN_STRATEGY_CLASSES:
        strategy = cls()
        backends = backend_map[strategy.name]
        result = strategy.assemble(request, backends, 10000, plan_context)
        if len(result) < 1:
            print(
                f"strategy-fail: {strategy.name} returned 0 fragments "
                "with populated backends"
            )
            all_ok = False
        else:
            print(f"strategy-ok: {strategy.name} returned {len(result)} fragment(s)")

    return 0 if all_ok else 1


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


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())