diff --git a/features/sliding_window_strategy.feature b/features/sliding_window_strategy.feature new file mode 100644 index 000000000..6ea28d745 --- /dev/null +++ b/features/sliding_window_strategy.feature @@ -0,0 +1,251 @@ +@phase2 @acms @context_strategies @sliding_window +Feature: SlidingWindowStrategy with configurable window size + As a CleverAgents developer + I want a SlidingWindowStrategy that implements the ContextStrategy protocol + So that long-running agent sessions can limit token usage by keeping only the most recent messages + + # =========================================================================== + # Protocol conformance + # =========================================================================== + + @sliding_window_protocol + Scenario: SlidingWindowStrategy satisfies the ContextStrategy protocol + Given a SlidingWindowStrategy with window_size 10 and window_mode "messages" + Then the sw strategy should satisfy the ContextStrategy protocol + And the sw strategy name should be "sliding_window" + And the sw strategy should have a capabilities object + + @sliding_window_protocol + Scenario: SlidingWindowStrategy explain returns a description + Given a SlidingWindowStrategy with window_size 5 and window_mode "messages" + Then the sw strategy explain should contain "sliding" + + # =========================================================================== + # Messages mode — basic truncation + # =========================================================================== + + @sliding_window_messages + Scenario: Messages mode keeps only the most recent N messages + Given a SlidingWindowStrategy with window_size 2 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 2 fragments should be returned by sliding window strategy + And the sliding window result should contain uko_node "project://app/b.py" + And the sliding window result should contain uko_node "project://app/c.py" + And the sliding window result should not contain uko_node "project://app/a.py" + + @sliding_window_messages + Scenario: Messages mode with exact boundary keeps all messages + Given a SlidingWindowStrategy with window_size 3 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 3 fragments should be returned by sliding window strategy + + @sliding_window_messages + Scenario: Messages mode with fewer messages than window size returns all + Given a SlidingWindowStrategy with window_size 10 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 2 fragments should be returned by sliding window strategy + + @sliding_window_messages + Scenario: Messages mode with empty context returns empty + Given a SlidingWindowStrategy with window_size 5 and window_mode "messages" + And an empty sliding window fragment list + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 0 fragments should be returned by sliding window strategy + + # =========================================================================== + # Tokens mode — basic truncation + # =========================================================================== + + @sliding_window_tokens + Scenario: Tokens mode keeps most recent messages within token limit + Given a SlidingWindowStrategy with window_size 25 and window_mode "tokens" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 2 fragments should be returned by sliding window strategy + And the sliding window result should contain uko_node "project://app/b.py" + And the sliding window result should contain uko_node "project://app/c.py" + And the sliding window result should not contain uko_node "project://app/a.py" + + @sliding_window_tokens + Scenario: Tokens mode with exact boundary keeps all messages + Given a SlidingWindowStrategy with window_size 30 and window_mode "tokens" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 3 fragments should be returned by sliding window strategy + + @sliding_window_tokens + Scenario: Tokens mode with empty context returns empty + Given a SlidingWindowStrategy with window_size 100 and window_mode "tokens" + And an empty sliding window fragment list + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 0 fragments should be returned by sliding window strategy + + @sliding_window_tokens + Scenario: Tokens mode with single large message that fits returns it + Given a SlidingWindowStrategy with window_size 50 and window_mode "tokens" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 40 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 1 fragments should be returned by sliding window strategy + + # =========================================================================== + # System prompt preservation + # =========================================================================== + + @sliding_window_system_prompt + Scenario: System prompt is always preserved outside the sliding window in messages mode + Given a SlidingWindowStrategy with window_size 2 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://system/prompt | system context | 0.9 | 20 | 5 | system | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 3 fragments should be returned by sliding window strategy + And the sliding window result should contain uko_node "project://system/prompt" + And the sliding window result should contain uko_node "project://app/b.py" + And the sliding window result should contain uko_node "project://app/c.py" + And the sliding window result should not contain uko_node "project://app/a.py" + + @sliding_window_system_prompt + Scenario: System prompt is always preserved outside the sliding window in tokens mode + Given a SlidingWindowStrategy with window_size 25 and window_mode "tokens" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://system/prompt | system context | 0.9 | 20 | 5 | system | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 3 fragments should be returned by sliding window strategy + And the sliding window result should contain uko_node "project://system/prompt" + And the sliding window result should contain uko_node "project://app/b.py" + And the sliding window result should contain uko_node "project://app/c.py" + And the sliding window result should not contain uko_node "project://app/a.py" + + @sliding_window_system_prompt + Scenario: Multiple system prompts are all preserved + Given a SlidingWindowStrategy with window_size 1 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://system/prompt1 | system context 1 | 0.9 | 20 | 5 | system | + | project://system/prompt2 | system context 2 | 0.8 | 20 | 5 | system | + | project://app/a.py | message 1 | 0.5 | 10 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 10 | 3 | user | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 3 fragments should be returned by sliding window strategy + And the sliding window result should contain uko_node "project://system/prompt1" + And the sliding window result should contain uko_node "project://system/prompt2" + And the sliding window result should contain uko_node "project://app/b.py" + And the sliding window result should not contain uko_node "project://app/a.py" + + @sliding_window_system_prompt + Scenario: Context with only system prompts returns all of them + Given a SlidingWindowStrategy with window_size 1 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://system/prompt1 | system context 1 | 0.9 | 20 | 5 | system | + | project://system/prompt2 | system context 2 | 0.8 | 20 | 5 | system | + And a sliding window budget with max_tokens 1000 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 2 fragments should be returned by sliding window strategy + + # =========================================================================== + # Budget enforcement + # =========================================================================== + + @sliding_window_budget + Scenario: Strategy respects token budget after sliding window truncation + Given a SlidingWindowStrategy with window_size 10 and window_mode "messages" + And the following sliding window fragments: + | uko_node | content | score | tokens | depth | role | + | project://app/a.py | message 1 | 0.5 | 100 | 3 | user | + | project://app/b.py | message 2 | 0.6 | 100 | 3 | user | + | project://app/c.py | message 3 | 0.7 | 100 | 3 | user | + And a sliding window budget with max_tokens 250 and reserved_tokens 0 + When I apply the SlidingWindowStrategy + Then 2 fragments should be returned by sliding window strategy + + # =========================================================================== + # can_handle + # =========================================================================== + + @sliding_window_can_handle + Scenario: can_handle returns a positive confidence + Given a SlidingWindowStrategy with window_size 10 and window_mode "messages" + When I check can_handle on SlidingWindowStrategy + Then the sliding window confidence should be greater than 0.0 + + # =========================================================================== + # Plugin registry registration + # =========================================================================== + + @sliding_window_registry + Scenario: SlidingWindowStrategy is registered under key "sliding_window" in the pipeline + Given an ACMS pipeline for sliding window tests + Then the sw pipeline should have strategy "sliding_window" + + @sliding_window_registry + Scenario: SlidingWindowStrategy can be registered manually with the pipeline + Given an ACMS pipeline for sliding window tests + When I register a custom SlidingWindowStrategy with the pipeline + Then the sw pipeline should have strategy "custom_sliding_window" + + # =========================================================================== + # Configuration validation + # =========================================================================== + + @sliding_window_config + Scenario: window_size must be positive + When I attempt to create a SlidingWindowStrategy with window_size 0 + Then a ValueError should be raised for sliding window + + @sliding_window_config + Scenario: window_mode must be messages or tokens + When I attempt to create a SlidingWindowStrategy with window_mode "invalid" + Then a ValueError should be raised for sliding window + + @sliding_window_config + Scenario: window_mode messages is valid + When I create a SlidingWindowStrategy with window_size 5 and window_mode "messages" + Then no error should be raised for sliding window + + @sliding_window_config + Scenario: window_mode tokens is valid + When I create a SlidingWindowStrategy with window_size 100 and window_mode "tokens" + Then no error should be raised for sliding window diff --git a/features/steps/sliding_window_strategy_steps.py b/features/steps/sliding_window_strategy_steps.py new file mode 100644 index 000000000..8d2d4ae22 --- /dev/null +++ b/features/steps/sliding_window_strategy_steps.py @@ -0,0 +1,262 @@ +"""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}" + ) diff --git a/src/cleveragents/application/services/sliding_window_strategy.py b/src/cleveragents/application/services/sliding_window_strategy.py new file mode 100644 index 000000000..20ae40386 --- /dev/null +++ b/src/cleveragents/application/services/sliding_window_strategy.py @@ -0,0 +1,262 @@ +"""SlidingWindowStrategy — configurable sliding window context strategy. + +Implements the ``ContextStrategy`` protocol for the ACMS pipeline. A +sliding window strategy limits token usage by keeping only the most +recent N messages or tokens in context, which is critical for +long-running agent sessions that would otherwise exceed LLM context +limits. + +The strategy supports two window modes: + +- **messages** — keeps the most recent ``window_size`` non-system + fragments. +- **tokens** — keeps the most recent fragments whose cumulative + ``token_count`` does not exceed ``window_size``. + +In both modes, fragments whose ``metadata["role"]`` is ``"system"`` +are always preserved outside the sliding window and are never subject +to truncation. + +The strategy is registered in the ``ACMSPipeline`` under the key +``"sliding_window"`` so it can be selected by name. + +Based on ``docs/specification.md`` §25207 and issue #9995. +""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from typing import Any, Literal + +from cleveragents.application.services.acms_service import ( + StrategyCapabilities, + _pack_budget, +) +from cleveragents.domain.models.core.context_fragment import ( + ContextBudget, + ContextFragment, +) + +logger = logging.getLogger(__name__) + +# Valid window mode literals. +WindowMode = Literal["messages", "tokens"] +_VALID_MODES: frozenset[str] = frozenset({"messages", "tokens"}) + +# Metadata key used to identify system-prompt fragments. +_ROLE_KEY: str = "role" +_SYSTEM_ROLE: str = "system" + + +def _is_system_fragment(fragment: ContextFragment) -> bool: + """Return True if *fragment* is a system-prompt fragment. + + A fragment is considered a system prompt when its ``metadata`` + contains ``{"role": "system"}``. + """ + return fragment.metadata.get(_ROLE_KEY) == _SYSTEM_ROLE + + +class SlidingWindowStrategy: + """Sliding window context strategy with configurable window size. + + Keeps only the most recent N messages or tokens in context, + preserving system-prompt fragments unconditionally. + + Parameters + ---------- + window_size: + Number of messages (``window_mode="messages"``) or maximum + cumulative token count (``window_mode="tokens"``) to retain. + Must be a positive integer. + window_mode: + ``"messages"`` — keep the most recent ``window_size`` non-system + fragments. + ``"tokens"`` — keep the most recent non-system fragments whose + cumulative ``token_count`` does not exceed ``window_size``. + + Raises + ------ + ValueError + If ``window_size`` is not a positive integer, or if + ``window_mode`` is not one of ``"messages"`` or ``"tokens"``. + + Implements ``ContextStrategy`` protocol. + """ + + def __init__( + self, + *, + window_size: int = 100, + window_mode: WindowMode = "messages", + ) -> None: + if window_size <= 0: + msg = ( + f"window_size must be a positive integer, got {window_size!r}" + ) + raise ValueError(msg) + if window_mode not in _VALID_MODES: + msg = ( + f"window_mode must be one of {sorted(_VALID_MODES)!r}, " + f"got {window_mode!r}" + ) + raise ValueError(msg) + self._window_size = window_size + self._window_mode: WindowMode = window_mode + + @property + def window_size(self) -> int: + """Return the configured window size.""" + return self._window_size + + @property + def window_mode(self) -> WindowMode: + """Return the configured window mode (``"messages"`` or ``"tokens"``).""" + return self._window_mode + + @property + def name(self) -> str: + return "sliding_window" + + @property + def capabilities(self) -> StrategyCapabilities: + return StrategyCapabilities( + supports_semantic_search=False, + supports_graph_navigation=False, + supports_temporal_archaeology=False, + quality_score=0.5, + ) + + def can_handle(self, request: dict[str, Any]) -> float: + """Return 0.5 confidence — general-purpose sliding window.""" + return 0.5 + + def assemble( + self, + fragments: Sequence[ContextFragment], + budget: ContextBudget, + ) -> Sequence[ContextFragment]: + """Apply the sliding window and return fragments within budget. + + Algorithm: + + 1. Separate *fragments* into system-prompt fragments (always + kept) and non-system fragments (subject to the window). + 2. Apply the sliding window to the non-system fragments, + keeping only the most recent ones that fit within + ``window_size``. + 3. Recombine system fragments with the windowed non-system + fragments, preserving the original relative order. + 4. Apply the token budget via ``_pack_budget``. + + Parameters + ---------- + fragments: + All context fragments for the current session, in + chronological order (oldest first). + budget: + Token budget for the assembled context. + + Returns + ------- + Sequence[ContextFragment] + The windowed and budget-constrained fragments. + """ + if not fragments: + return [] + + # Partition into system and non-system fragments, preserving order. + system_frags: list[ContextFragment] = [] + non_system_frags: list[ContextFragment] = [] + for frag in fragments: + if _is_system_fragment(frag): + system_frags.append(frag) + else: + non_system_frags.append(frag) + + # Apply the sliding window to non-system fragments. + windowed_non_system = self._apply_window(non_system_frags) + + # Rebuild the combined list preserving original order. + # We use the original fragment list as the ordering reference. + windowed_ids: set[str] = {f.fragment_id for f in windowed_non_system} + system_ids: set[str] = {f.fragment_id for f in system_frags} + combined: list[ContextFragment] = [ + f + for f in fragments + if f.fragment_id in system_ids or f.fragment_id in windowed_ids + ] + + logger.info( + "SlidingWindow assembled context", + extra={ + "window_size": self._window_size, + "window_mode": self._window_mode, + "total_input": len(fragments), + "system_count": len(system_frags), + "windowed_count": len(windowed_non_system), + "combined_count": len(combined), + }, + ) + + return _pack_budget(combined, budget) + + def _apply_window( + self, + non_system_frags: list[ContextFragment], + ) -> list[ContextFragment]: + """Apply the sliding window to *non_system_frags*. + + Returns the most recent fragments that fit within the window, + in their original (chronological) order. + """ + if not non_system_frags: + return [] + + if self._window_mode == "messages": + return self._apply_message_window(non_system_frags) + # window_mode == "tokens" + return self._apply_token_window(non_system_frags) + + def _apply_message_window( + self, + frags: list[ContextFragment], + ) -> list[ContextFragment]: + """Keep the most recent ``window_size`` fragments.""" + if len(frags) <= self._window_size: + return frags + # Slice from the end to keep the most recent N. + return frags[-self._window_size :] + + def _apply_token_window( + self, + frags: list[ContextFragment], + ) -> list[ContextFragment]: + """Keep the most recent fragments within the token budget. + + Iterates from the most recent fragment backwards, accumulating + token counts until the window is full. Returns the selected + fragments in their original chronological order. + """ + selected: list[ContextFragment] = [] + accumulated = 0 + for frag in reversed(frags): + if accumulated + frag.token_count <= self._window_size: + selected.append(frag) + accumulated += frag.token_count + else: + break + # Reverse to restore chronological order. + selected.reverse() + return selected + + def explain(self) -> str: + return ( + f"Sliding window strategy ({self._window_mode} mode, " + f"window_size={self._window_size}). " + "Keeps only the most recent messages or tokens in context, " + "preserving system-prompt fragments unconditionally. " + "Critical for long-running agent sessions that would otherwise " + "exceed LLM context limits." + )