cleveragents-core/features/steps/sliding_window_strategy_steps.py

"""Step definitions for ``features/sliding_window_strategy.feature``.

Covers the SlidingWindowStrategy:

* Protocol conformance
* Messages mode truncation
* Tokens mode truncation
* System prompt preservation
* Budget enforcement
* can_handle confidence
* Plugin registry registration
* Configuration validation
"""

from __future__ import annotations

from typing import Any

from behave import given, then, when
from behave.runner import Context

from cleveragents.application.services.acms_service import (
    ACMSPipeline,
    ContextStrategy,
)
from cleveragents.application.services.sliding_window_strategy import (
    SlidingWindowStrategy,
)
from cleveragents.domain.models.core.context_fragment import (
    ContextBudget,
    ContextFragment,
    FragmentProvenance,
)

__all__: list[str] = []


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


def _make_sw_fragment(
    uko_node: str,
    content: str,
    score: float,
    tokens: int,
    depth: int,
    role: str = "user",
) -> ContextFragment:
    """Build a ``ContextFragment`` for sliding window tests."""
    metadata: dict[str, str] = {}
    if role:
        metadata["role"] = role
    return ContextFragment(
        uko_node=uko_node,
        content=content,
        relevance_score=score,
        token_count=tokens,
        detail_depth=depth,
        provenance=FragmentProvenance(resource_uri=uko_node),
        metadata=metadata,
    )


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given('a SlidingWindowStrategy with window_size {size:d} and window_mode "{mode}"')
def step_sliding_window_strategy(context: Context, size: int, mode: str) -> None:
    context.sw_strategy = SlidingWindowStrategy(
        window_size=size,
        window_mode=mode,  # type: ignore[arg-type]
    )


@given("the following sliding window fragments:")
def step_sw_fragments_table(context: Context) -> None:
    context.sw_fragments = []
    for row in context.table:
        role = row["role"] if "role" in row.headings else "user"
        frag = _make_sw_fragment(
            uko_node=row["uko_node"],
            content=row["content"],
            score=float(row["score"]),
            tokens=int(row["tokens"]),
            depth=int(row["depth"]),
            role=role,
        )
        context.sw_fragments.append(frag)


@given("an empty sliding window fragment list")
def step_sw_empty_fragments(context: Context) -> None:
    context.sw_fragments = []


@given(
    "a sliding window budget with max_tokens {max_tokens:d} and reserved_tokens {reserved:d}"
)
def step_sw_budget(context: Context, max_tokens: int, reserved: int) -> None:
    context.sw_budget = ContextBudget(max_tokens=max_tokens, reserved_tokens=reserved)


@given("an ACMS pipeline for sliding window tests")
def step_sw_pipeline(context: Context) -> None:
    # Create the pipeline and register SlidingWindowStrategy under "sliding_window".
    # This satisfies the requirement to register the strategy in the plugin
    # registry under key "sliding_window".
    pipeline = ACMSPipeline()
    pipeline.register_strategy("sliding_window", SlidingWindowStrategy())
    context.sw_pipeline = pipeline


# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------


@when("I apply the SlidingWindowStrategy")
def step_apply_sw(context: Context) -> None:
    context.sw_result = list(
        context.sw_strategy.assemble(context.sw_fragments, context.sw_budget)
    )


@when("I check can_handle on SlidingWindowStrategy")
def step_sw_can_handle(context: Context) -> None:
    request: dict[str, Any] = {}
    context.sw_confidence = context.sw_strategy.can_handle(request)


@when("I register a custom SlidingWindowStrategy with the pipeline")
def step_register_custom_sw(context: Context) -> None:
    custom_strategy = SlidingWindowStrategy(window_size=50, window_mode="messages")
    context.sw_pipeline.register_strategy("custom_sliding_window", custom_strategy)


@when("I attempt to create a SlidingWindowStrategy with window_size {size:d}")
def step_attempt_create_sw_bad_size(context: Context, size: int) -> None:
    context.sw_error = None
    try:
        SlidingWindowStrategy(window_size=size, window_mode="messages")
    except ValueError as exc:
        context.sw_error = exc


@when('I attempt to create a SlidingWindowStrategy with window_mode "{mode}"')
def step_attempt_create_sw_bad_mode(context: Context, mode: str) -> None:
    context.sw_error = None
    try:
        SlidingWindowStrategy(window_size=10, window_mode=mode)  # type: ignore[arg-type]
    except ValueError as exc:
        context.sw_error = exc


@when(
    'I create a SlidingWindowStrategy with window_size {size:d} and window_mode "{mode}"'
)
def step_create_sw_valid(context: Context, size: int, mode: str) -> None:
    context.sw_error = None
    try:
        context.sw_strategy = SlidingWindowStrategy(
            window_size=size,
            window_mode=mode,  # type: ignore[arg-type]
        )
    except ValueError as exc:
        context.sw_error = exc


# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------


@then("the sw strategy should satisfy the ContextStrategy protocol")
def step_sw_satisfies_protocol(context: Context) -> None:
    assert isinstance(context.sw_strategy, ContextStrategy), (
        f"SlidingWindowStrategy does not satisfy ContextStrategy protocol: "
        f"{type(context.sw_strategy)}"
    )


@then('the sw strategy name should be "sliding_window"')
def step_sw_name_sliding_window(context: Context) -> None:
    actual = context.sw_strategy.name
    assert actual == "sliding_window", f"Expected name 'sliding_window', got '{actual}'"


@then("the sw strategy should have a capabilities object")
def step_sw_has_capabilities(context: Context) -> None:
    caps = context.sw_strategy.capabilities
    assert caps is not None, "Expected capabilities object, got None"


@then('the sw strategy explain should contain "{text}"')
def step_sw_explain_contains(context: Context, text: str) -> None:
    explanation = context.sw_strategy.explain()
    assert text.lower() in explanation.lower(), (
        f"Expected explain to contain '{text}', got: {explanation}"
    )


@then("{count:d} fragments should be returned by sliding window strategy")
def step_sw_fragment_count(context: Context, count: int) -> None:
    actual = len(context.sw_result)
    assert actual == count, (
        f"Expected {count} fragments from sliding window, got {actual}. "
        f"Fragments: {[f.uko_node for f in context.sw_result]}"
    )


@then('the sliding window result should contain uko_node "{uko_node}"')
def step_sw_result_contains(context: Context, uko_node: str) -> None:
    nodes = [f.uko_node for f in context.sw_result]
    assert uko_node in nodes, f"Expected uko_node '{uko_node}' in result, got: {nodes}"


@then('the sliding window result should not contain uko_node "{uko_node}"')
def step_sw_result_not_contains(context: Context, uko_node: str) -> None:
    nodes = [f.uko_node for f in context.sw_result]
    assert uko_node not in nodes, (
        f"Expected uko_node '{uko_node}' NOT in result, but it was. Nodes: {nodes}"
    )


@then("the sliding window confidence should be greater than 0.0")
def step_sw_confidence_positive(context: Context) -> None:
    actual = context.sw_confidence
    assert actual > 0.0, f"Expected confidence > 0.0, got {actual}"


@then('the sw pipeline should have strategy "{name}"')
def step_sw_pipeline_has_strategy(context: Context, name: str) -> None:
    registered = context.sw_pipeline._strategies
    assert name in registered, (
        f"Pipeline does not have strategy '{name}'. "
        f"Registered: {list(registered.keys())}"
    )


@then("a ValueError should be raised for sliding window")
def step_sw_value_error_raised(context: Context) -> None:
    assert context.sw_error is not None, (
        "Expected a ValueError to be raised, but no error was raised"
    )
    assert isinstance(context.sw_error, ValueError), (
        f"Expected ValueError, got {type(context.sw_error)}: {context.sw_error}"
    )


@then("no error should be raised for sliding window")
def step_sw_no_error(context: Context) -> None:
    assert context.sw_error is None, f"Expected no error, but got: {context.sw_error}"