cleveragents-core/robot/helper_context_fragment_models.py

"""Robot Framework helper for ScoredFragment and ContextFragment integration tests.

Provides a CLI-style interface for Robot to invoke pipeline model
operations and verify results.  Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_context_fragment_models.py scored-create
    python robot/helper_context_fragment_models.py scored-breakdown
    python robot/helper_context_fragment_models.py scored-equality
    python robot/helper_context_fragment_models.py scored-dedup
    python robot/helper_context_fragment_models.py scored-from-relevance
    python robot/helper_context_fragment_models.py scored-frozen
    python robot/helper_context_fragment_models.py strategy-source
"""

from __future__ import annotations

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

from pydantic import ValidationError

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

from cleveragents.domain.contexts.fragment import ScoredFragment  # noqa: E402
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextFragment,
    FragmentProvenance,
)

_DEFAULT_PROV = FragmentProvenance(resource_uri="test://robot")


def _make_fragment(**kwargs: object) -> ContextFragment:
    """Create a ContextFragment with sensible test defaults."""
    defaults: dict[str, object] = {
        "uko_node": "test://default",
        "content": "test content",
        "token_count": 10,
        "provenance": _DEFAULT_PROV,
    }
    defaults.update(kwargs)
    return ContextFragment(**defaults)  # type: ignore[arg-type]


def _cmd_scored_create() -> int:
    """Create a ScoredFragment and verify defaults."""
    frag = _make_fragment(uko_node="project://app/main.py", content="hello")
    scored = ScoredFragment(fragment=frag, composite_score=0.85)
    if scored.composite_score != 0.85:
        print(f"fail: expected composite_score=0.85, got {scored.composite_score}")
        return 1
    if scored.rank != 0:
        print(f"fail: expected rank=0, got {scored.rank}")
        return 1
    if scored.score_breakdown != {}:
        print(f"fail: expected empty breakdown, got {scored.score_breakdown}")
        return 1
    if scored.uko_node != "project://app/main.py":
        print(f"fail: expected uko_node=project://app/main.py, got {scored.uko_node}")
        return 1
    print("scored-create-ok")
    return 0


def _cmd_scored_breakdown() -> int:
    """Create a ScoredFragment with full breakdown."""
    frag = _make_fragment(uko_node="project://app/io.py", content="async IO")
    breakdown = {"relevance": 0.9, "hierarchy": 0.8, "recency": 0.95}
    scored = ScoredFragment(
        fragment=frag,
        composite_score=0.92,
        score_breakdown=breakdown,
        rank=1,
    )
    if scored.composite_score != 0.92:
        print(f"fail: expected composite_score=0.92, got {scored.composite_score}")
        return 1
    if scored.rank != 1:
        print(f"fail: expected rank=1, got {scored.rank}")
        return 1
    if scored.score_breakdown.get("relevance") != 0.9:
        print(f"fail: expected relevance=0.9, got {scored.score_breakdown}")
        return 1
    print("scored-breakdown-ok")
    return 0


def _cmd_scored_equality() -> int:
    """Verify equality based on uko_node + detail_depth."""
    frag_a = _make_fragment(
        uko_node="project://app/main.py",
        content="A",
        detail_depth=3,
    )
    frag_b = _make_fragment(
        uko_node="project://app/main.py",
        content="B",
        detail_depth=3,
    )
    sa = ScoredFragment.from_fragment(frag_a, composite_score=0.5)
    sb = ScoredFragment.from_fragment(frag_b, composite_score=0.9)
    if sa != sb:
        print("fail: expected equal scored fragments")
        return 1
    if hash(sa) != hash(sb):
        print("fail: expected same hash")
        return 1
    # Different uko_node should be not equal
    frag_c = _make_fragment(
        uko_node="project://app/other.py",
        content="C",
        detail_depth=3,
    )
    sc = ScoredFragment.from_fragment(frag_c, composite_score=0.5)
    if sa == sc:
        print("fail: expected different scored fragments")
        return 1
    print("scored-equality-ok")
    return 0


def _cmd_scored_dedup() -> int:
    """Verify set-based deduplication."""
    frag_a = _make_fragment(
        uko_node="project://app/main.py",
        content="A",
        detail_depth=3,
    )
    frag_b = _make_fragment(
        uko_node="project://app/main.py",
        content="B",
        detail_depth=3,
    )
    frag_c = _make_fragment(
        uko_node="project://app/other.py",
        content="C",
        detail_depth=3,
    )
    scored_set = {
        ScoredFragment.from_fragment(frag_a, composite_score=0.5),
        ScoredFragment.from_fragment(frag_b, composite_score=0.9),
        ScoredFragment.from_fragment(frag_c, composite_score=0.7),
    }
    if len(scored_set) != 2:
        print(f"fail: expected 2 unique, got {len(scored_set)}")
        return 1
    print("scored-dedup-ok")
    return 0


def _cmd_scored_from_relevance() -> int:
    """Create ScoredFragment using from_relevance factory."""
    frag = _make_fragment(
        uko_node="project://app/util.py",
        content="utils",
        relevance_score=0.75,
    )
    scored = ScoredFragment.from_relevance(frag)
    if scored.composite_score != 0.75:
        print(f"fail: expected composite_score=0.75, got {scored.composite_score}")
        return 1
    if scored.score_breakdown.get("relevance") != 0.75:
        print(f"fail: expected relevance=0.75, got {scored.score_breakdown}")
        return 1
    print("scored-from-relevance-ok")
    return 0


def _cmd_scored_frozen() -> int:
    """Verify ScoredFragment is frozen."""
    frag = _make_fragment()
    scored = ScoredFragment(fragment=frag, composite_score=0.5)
    try:
        scored.composite_score = 0.99  # type: ignore[misc]
        print("fail: expected error when modifying frozen model")
        return 1
    except ValidationError:
        pass
    print("scored-frozen-ok")
    return 0


def _cmd_strategy_source() -> int:
    """Verify strategy_source field on ContextFragment."""
    # Default should be empty string
    frag = _make_fragment()
    if frag.strategy_source != "":
        print(f"fail: expected empty strategy_source, got '{frag.strategy_source}'")
        return 1
    # Explicit value
    frag2 = _make_fragment(strategy_source="keyword-search")
    if frag2.strategy_source != "keyword-search":
        print(
            f"fail: expected strategy_source='keyword-search', "
            f"got '{frag2.strategy_source}'"
        )
        return 1
    print("strategy-source-ok")
    return 0


_COMMANDS: dict[str, Callable[[], int]] = {
    "scored-create": _cmd_scored_create,
    "scored-breakdown": _cmd_scored_breakdown,
    "scored-equality": _cmd_scored_equality,
    "scored-dedup": _cmd_scored_dedup,
    "scored-from-relevance": _cmd_scored_from_relevance,
    "scored-frozen": _cmd_scored_frozen,
    "strategy-source": _cmd_strategy_source,
}


def main() -> int:
    """Dispatch to the requested sub-command."""
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        valid = ", ".join(sorted(_COMMANDS))
        print(f"Usage: {sys.argv[0]} <{valid}>", file=sys.stderr)
        return 1
    return _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    raise SystemExit(main())