diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f086d79c..9defc8736 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## [Unreleased] +### Fixed + +- Attachment turns in long sessions can now use one bounded automatic history + compaction when removable history is the only known capacity pressure, while + the current turn and its attachments remain protected. Deployments with + unknown context limits and Ensemble routes continue to fail closed with + actionable guidance. + ## [0.5.4] - 2026-08-25 ### Added diff --git a/src/opensquilla/engine/capacity_admission.py b/src/opensquilla/engine/capacity_admission.py index 5cf37b524..669f6f1bf 100644 --- a/src/opensquilla/engine/capacity_admission.py +++ b/src/opensquilla/engine/capacity_admission.py @@ -2,6 +2,9 @@ from __future__ import annotations +from dataclasses import dataclass +from typing import Literal + from opensquilla.context_budget import CHARS_PER_TOKEN, ContextBudgetGovernor from opensquilla.provider.model_catalog import ( resolve_effective_context_window, @@ -14,13 +17,54 @@ "For a custom or catalog-unknown model, set llm.context_window_tokens " "to the deployment's verified context limit." ) +CAPACITY_REDUCTION_HINT = ( + "Reduce the attachment or session context, run /compact, or start a new " + "session before retrying." +) +CAPACITY_TOO_LARGE_ERROR_CODE = "attachment_capacity_too_large" +CAPACITY_UNKNOWN_ERROR_CODE = "attachment_capacity_unknown" + +CapacityAdmissionStatus = Literal[ + "fits", + "known_capacity_request_too_large", + "capacity_unknown", +] class LargeContextCapacityError(RuntimeError): """An attachment turn has no deployment with proven request capacity.""" + def __init__( + self, + message: str, + *, + status: CapacityAdmissionStatus | None = None, + ) -> None: + super().__init__(message) + self.status = status + self.code = ( + CAPACITY_TOO_LARGE_ERROR_CODE + if status == "known_capacity_request_too_large" + else CAPACITY_UNKNOWN_ERROR_CODE + if status == "capacity_unknown" + else "attachment_capacity_unavailable" + ) -def model_has_request_capacity( + +@dataclass(frozen=True, slots=True) +class ModelRequestCapacityAssessment: + """Structured proof result for one physical model deployment.""" + + status: CapacityAdmissionStatus + required_input_tokens: int + safe_input_tokens: int | None + + @property + def fits(self) -> bool: + return self.status == "fits" + + +def assess_model_request_capacity( *, provider: str, model: str, @@ -33,23 +77,30 @@ def model_has_request_capacity( api_key: str = "", base_url: str = "", proxy: str = "", -) -> bool: - """Return whether definite catalog limits prove a conservative request fits. - - ``request_input_tokens`` is the preferred path: callers that can measure the - assembled request pass the complete input estimate. ``material_tokens`` plus - the historical fixed reserve remains only for compatibility with older - internal callers that have not reached an assembled-request boundary. - """ +) -> ModelRequestCapacityAssessment: + """Classify whether definite deployment limits prove a request fits.""" provider_id = str(provider or "").strip() model_id = str(model or "").strip() resolved_request_tokens = max(0, int(request_input_tokens)) resolved_material_tokens = max(0, int(material_tokens)) - if not provider_id or not model_id or ( - resolved_request_tokens <= 0 and resolved_material_tokens <= 0 - ): - return False + required_input_tokens = ( + resolved_request_tokens + if resolved_request_tokens > 0 + else ( + resolved_material_tokens + NON_MATERIAL_INPUT_HEADROOM_TOKENS + if resolved_material_tokens > 0 + else 0 + ) + ) + unknown = ModelRequestCapacityAssessment( + status="capacity_unknown", + required_input_tokens=required_input_tokens, + safe_input_tokens=None, + ) + if not provider_id or not model_id or required_input_tokens <= 0: + return unknown + catalog = shared_catalog() try: window, window_source = resolve_effective_context_window( @@ -82,34 +133,81 @@ def model_has_request_capacity( max_output, int(deployment_limits.max_output_tokens), ) + if window_source not in {"catalog", "config", "override"}: + return unknown + budget = ContextBudgetGovernor.from_values( + context_window_tokens=window, + max_output_tokens=max_output, + thinking_budget_tokens=max(0, int(thinking_budget_tokens)), + context_overflow_threshold=0.85, + ).snapshot() + safe_input_tokens = budget.provider_request_max_chars // CHARS_PER_TOKEN + if provider_request_proof_max_chars > 0: + safe_input_tokens = min( + safe_input_tokens, + int(provider_request_proof_max_chars) // CHARS_PER_TOKEN, + ) except Exception: # noqa: BLE001 - invalid/missing capability fails closed - return False - if window_source not in {"catalog", "config", "override"}: - return False - budget = ContextBudgetGovernor.from_values( - context_window_tokens=window, - max_output_tokens=max_output, - thinking_budget_tokens=max(0, int(thinking_budget_tokens)), - context_overflow_threshold=0.85, - ).snapshot() - safe_input_tokens = budget.provider_request_max_chars // CHARS_PER_TOKEN - if provider_request_proof_max_chars > 0: - safe_input_tokens = min( - safe_input_tokens, - int(provider_request_proof_max_chars) // CHARS_PER_TOKEN, - ) - required_input_tokens = ( - resolved_request_tokens - if resolved_request_tokens > 0 - else resolved_material_tokens + NON_MATERIAL_INPUT_HEADROOM_TOKENS + return unknown + + return ModelRequestCapacityAssessment( + status=( + "fits" + if required_input_tokens <= safe_input_tokens + else "known_capacity_request_too_large" + ), + required_input_tokens=required_input_tokens, + safe_input_tokens=safe_input_tokens, ) - return required_input_tokens <= safe_input_tokens + + +def model_has_request_capacity( + *, + provider: str, + model: str, + material_tokens: int, + thinking_budget_tokens: int, + request_input_tokens: int = 0, + context_window_override_tokens: int = 0, + max_output_override_tokens: int = 0, + provider_request_proof_max_chars: int = 0, + api_key: str = "", + base_url: str = "", + proxy: str = "", +) -> bool: + """Return whether definite catalog limits prove a conservative request fits. + + ``request_input_tokens`` is the preferred path: callers that can measure the + assembled request pass the complete input estimate. ``material_tokens`` plus + the historical fixed reserve remains only for compatibility with older + internal callers that have not reached an assembled-request boundary. + """ + + return assess_model_request_capacity( + provider=provider, + model=model, + material_tokens=material_tokens, + thinking_budget_tokens=thinking_budget_tokens, + request_input_tokens=request_input_tokens, + context_window_override_tokens=context_window_override_tokens, + max_output_override_tokens=max_output_override_tokens, + provider_request_proof_max_chars=provider_request_proof_max_chars, + api_key=api_key, + base_url=base_url, + proxy=proxy, + ).fits __all__ = [ "CAPACITY_CONFIGURATION_HINT", + "CAPACITY_REDUCTION_HINT", + "CAPACITY_TOO_LARGE_ERROR_CODE", + "CAPACITY_UNKNOWN_ERROR_CODE", + "CapacityAdmissionStatus", "LargeContextCapacityError", "MAX_THINKING_BUDGET_TOKENS", + "ModelRequestCapacityAssessment", "NON_MATERIAL_INPUT_HEADROOM_TOKENS", + "assess_model_request_capacity", "model_has_request_capacity", ] diff --git a/src/opensquilla/engine/runtime.py b/src/opensquilla/engine/runtime.py index 6d972f6f2..12f29a3ad 100644 --- a/src/opensquilla/engine/runtime.py +++ b/src/opensquilla/engine/runtime.py @@ -5487,6 +5487,11 @@ async def _load_turn_transcript() -> Sequence[Any]: # or ensemble wrapping changes ``provider`` for this one turn. durable_base_consumer_provider = provider cloned_selector = pt_out.cloned_selector + pre_router_provider_config = ( + getattr(cloned_selector, "current_config", None) + if cloned_selector is not None + else None + ) tool_defs = pt_out.tool_defs tool_handler = pt_out.tool_handler tool_context = pt_out.effective_tool_context @@ -6000,8 +6005,16 @@ def _refresh_compaction_plan_for_operation() -> Any | None: stable_consumer_proof_max_chars = ( agent.config.provider_request_proof_max_chars ) + capacity_retry_pending = bool( + turn.metadata.get("large_context_capacity_retry_pending") is True + ) + compaction_consumer_provider = ( + provider + if capacity_retry_pending + else durable_base_consumer_provider + ) stable_consumer_metadata = provider_metadata( - durable_base_consumer_provider + compaction_consumer_provider ) llm_cfg = ( getattr(self._config, "llm", None) @@ -6011,6 +6024,7 @@ def _refresh_compaction_plan_for_operation() -> Any | None: if ( bool(turn.metadata.get("routing_applied", False)) and self._model_catalog is not None + and not capacity_retry_pending ): ( base_provider, @@ -6064,7 +6078,10 @@ def _refresh_compaction_plan_for_operation() -> Any | None: stable_compaction_window_tokens = _durable_compaction_window_tokens( compaction_context_window_tokens, stable_consumer_window_tokens=stable_consumer_window_tokens, - routing_applied=bool(turn.metadata.get("routing_applied", False)), + routing_applied=bool( + turn.metadata.get("routing_applied", False) + and not capacity_retry_pending + ), ) if stable_compaction_window_tokens != compaction_context_window_tokens: log.info( @@ -6080,7 +6097,7 @@ def _refresh_compaction_plan_for_operation() -> Any | None: ) if callable(bind_durable_consumer): bind_durable_consumer( - provider=durable_base_consumer_provider, + provider=compaction_consumer_provider, model_id=stable_consumer_model_id, context_window_tokens=stable_consumer_window_tokens, max_output_tokens=stable_consumer_max_output_tokens, @@ -6122,7 +6139,7 @@ def _refresh_compaction_plan_for_operation() -> Any | None: attachments=attachments, attachment_messages=extra_msgs, context_window_tokens=compaction_context_window_tokens, - consumer_provider=durable_base_consumer_provider, + consumer_provider=compaction_consumer_provider, consumer_max_output_tokens=( stable_consumer_max_output_tokens ), @@ -6136,7 +6153,7 @@ def _refresh_compaction_plan_for_operation() -> Any | None: consumer_admission, consumer_admission_fingerprint, ) = build_consumer_admission( - consumer_provider=durable_base_consumer_provider, + consumer_provider=compaction_consumer_provider, active_user_message=effective_runtime_message, active_user_in_history=history_has_persisted_user, bound_user_message_id=bound_user_message_id, @@ -6201,6 +6218,81 @@ def _refresh_compaction_plan_for_operation() -> Any | None: ) ch_out = ch_outcome.require_output() agent.config.request_context_prompt = ch_out.final_request_context_prompt + if turn.metadata.get("large_context_capacity_retry_pending") is True: + turn.metadata["large_context_capacity_compaction_t3_status"] = ( + ch_out.t3_upgrade_status + ) + turn.metadata["large_context_capacity_compaction_preflight_invoked"] = ( + ch_out.preflight_invoked + ) + retry_history_capacity = ( + await self._router_history_capacity_for_request( + session_key, + RouterHistoryReplayRequest( + exclude_last_user=( + history_has_persisted_user or persist_input + ), + bound_user_message_id=bound_user_message_id, + transcript_snapshot=transcript_snapshot, + ), + max_history_turns=self._route_history_turn_limit( + turn.metadata + ), + preserve_image_attachments=( + turn.metadata.get("image_route_reason") + in {"current_turn", "gate_history"} + ), + reachable_provider_kinds=( + self._route_capacity_provider_kinds( + turn, + initial_provider_config=pre_router_provider_config, + ) + ), + ) + ) + turn.metadata["routing_history_capacity_estimated_tokens"] = max( + 0, + int( + retry_history_capacity.get( + "history_capacity_estimated_tokens" + ) + or 0 + ), + ) + turn.metadata["routing_history_capacity_message_count"] = max( + 0, + int( + retry_history_capacity.get( + "history_capacity_message_count" + ) + or 0 + ), + ) + turn.metadata["routing_history_capacity_estimate_complete"] = ( + retry_history_capacity.get( + "history_capacity_estimate_complete" + ) + is True + ) + from opensquilla.engine.selector_override import ( + require_current_selector_capacity, + ) + from opensquilla.engine.steps.squilla_router import ( + finalize_squilla_router_capacity, + ) + + turn = await finalize_squilla_router_capacity( + turn, + retry_after_compaction=True, + ) + require_current_selector_capacity( + cloned_selector, + turn.metadata, + reason=( + "The final model deployment does not have proven capacity " + "for this attachment request after automatic compaction." + ), + ) compaction_source_entries: tuple[Any, ...] | None = None compaction_source_preimage: tuple[tuple[Any, ...], ...] | None = None @@ -6855,6 +6947,10 @@ def _refresh_compaction_plan_for_operation() -> Any | None: fallback_error_class="agent_error", fallback_error_message=str(exc) or "Agent error", ) + from opensquilla.engine.capacity_admission import ( + LargeContextCapacityError, + ) + if provider_boundary_failure_kind: event_code = safe_provider_failure_code( str(getattr(exc, "code", "") or ""), @@ -6864,6 +6960,13 @@ def _refresh_compaction_plan_for_operation() -> Any | None: error_message = safe_provider_failure_message( provider_boundary_failure_kind ) + elif isinstance(exc, LargeContextCapacityError): + event_code = str( + getattr(exc, "code", "attachment_capacity_unavailable") + or "attachment_capacity_unavailable" + ) + error_code = event_code + error_message = str(exc) elif isinstance(exc, UsageAccountingUnavailableError): event_code = str( getattr(exc, "code", UsageAccountingUnavailableError.code) @@ -9409,7 +9512,36 @@ def _run_router_step_sync() -> TurnContext: # prompt/tool boundary outside the generic fail-open pipeline wrapper. # An unexpected estimator failure must stop the turn rather than leave # an attachment route with unbounded selector fallbacks. - turn = await finalize_squilla_router_capacity(turn) + turn = await finalize_squilla_router_capacity( + turn, + allow_compaction_retry=bool( + router_history_replay_request is not None + and cloned_selector is not None + and self._session_manager is not None + and not bool( + getattr( + getattr(turn.config, "llm_ensemble", None), + "enabled", + False, + ) + ) + ), + ) + if turn.metadata.get("large_context_capacity_blocked") is True: + from opensquilla.engine.selector_override import ( + require_current_selector_capacity, + ) + + # Fixed-baseline Ensemble paths intentionally skip the normal + # model-override call below. Enforce the finalized attachment + # capacity block before any such path can reach a provider. + require_current_selector_capacity( + cloned_selector, + turn.metadata, + reason=( + "No deployment has proven capacity for this attachment request." + ), + ) # Image routing is a capability boundary, not an Ensemble activation. # This applies to the dedicated image row and to any text tier selected diff --git a/src/opensquilla/engine/selector_override.py b/src/opensquilla/engine/selector_override.py index a1d5dcca0..40856b565 100644 --- a/src/opensquilla/engine/selector_override.py +++ b/src/opensquilla/engine/selector_override.py @@ -58,26 +58,24 @@ def _bounded_fallback_chain_required(turn_metadata: dict[str, Any]) -> bool: ) -def _provider_config_has_request_capacity( +def _provider_config_capacity_assessment( config: Any, turn_metadata: dict[str, Any], *, provider: str = "", model: str = "", -) -> bool: - """Validate one final physical deployment against the routed material.""" +) -> Any: + """Return the structured request-capacity proof for one deployment.""" - if not _large_context_capacity_required(turn_metadata): - return True from opensquilla.engine.capacity_admission import ( MAX_THINKING_BUDGET_TOKENS, - model_has_request_capacity, + assess_model_request_capacity, ) thinking_budget = turn_metadata.get("large_context_thinking_budget_tokens") if not isinstance(thinking_budget, int) or isinstance(thinking_budget, bool): thinking_budget = MAX_THINKING_BUDGET_TOKENS - return model_has_request_capacity( + return assess_model_request_capacity( provider=( str(getattr(config, "provider", "") or "").strip() or str(provider or "").strip() @@ -113,6 +111,84 @@ def _provider_config_has_request_capacity( ) +def _provider_config_has_request_capacity( + config: Any, + turn_metadata: dict[str, Any], + *, + provider: str = "", + model: str = "", +) -> bool: + """Validate one final physical deployment against the routed material.""" + + if not _large_context_capacity_required(turn_metadata): + return True + return _provider_config_capacity_assessment( + config, + turn_metadata, + provider=provider, + model=model, + ).fits + + +def _provisional_capacity_binding_allowed( + config: Any, + turn_metadata: dict[str, Any], + assessment: Any, + *, + provider: str, + model: str, +) -> bool: + """Permit only the exact known deployment needed to reach compaction.""" + + if ( + turn_metadata.get("large_context_capacity_retry_pending") is not True + or turn_metadata.get("large_context_capacity_retry_attempted") is True + or getattr(assessment, "status", None) + != "known_capacity_request_too_large" + ): + return False + actual_provider = ( + str(getattr(config, "provider", "") or "").strip() + or str(provider or "").strip() + ).lower() + actual_model = ( + str(getattr(config, "model", "") or "").strip() + or str(model or "").strip() + ) + expected_provider = str( + turn_metadata.get("large_context_capacity_provisional_provider") or "" + ).strip().lower() + expected_model = str( + turn_metadata.get("large_context_capacity_provisional_model") or "" + ).strip() + if ( + not expected_provider + or not expected_model + or actual_provider != expected_provider + or actual_model != expected_model + ): + return False + safe_input_tokens = getattr(assessment, "safe_input_tokens", None) + if not isinstance(safe_input_tokens, int) or isinstance(safe_input_tokens, bool): + return False + request_input_tokens = _metadata_nonnegative_int( + turn_metadata, + "large_context_request_input_tokens", + ) + history_tokens = _metadata_nonnegative_int( + turn_metadata, + "large_context_history_tokens", + ) + if history_tokens <= 0 or request_input_tokens - history_tokens > safe_input_tokens: + return False + turn_metadata["large_context_capacity_provisional_bound"] = True + turn_metadata["large_context_capacity_provisional_physical_provider"] = ( + actual_provider + ) + turn_metadata["large_context_capacity_provisional_physical_model"] = actual_model + return True + + def _require_provider_config_capacity( config: Any, turn_metadata: dict[str, Any], @@ -121,22 +197,63 @@ def _require_provider_config_capacity( provider: str = "", model: str = "", ) -> None: - if _provider_config_has_request_capacity( + if not _large_context_capacity_required(turn_metadata): + return + assessment = _provider_config_capacity_assessment( config, turn_metadata, provider=provider, model=model, + ) + if assessment.fits or _provisional_capacity_binding_allowed( + config, + turn_metadata, + assessment, + provider=provider, + model=model, ): return from opensquilla.engine.capacity_admission import ( CAPACITY_CONFIGURATION_HINT, + CAPACITY_REDUCTION_HINT, LargeContextCapacityError, ) turn_metadata["large_context_capacity_blocked"] = True - actionable_reason = f"{reason} {CAPACITY_CONFIGURATION_HINT}" + turn_metadata["large_context_capacity_status"] = assessment.status + hint = ( + CAPACITY_REDUCTION_HINT + if assessment.status == "known_capacity_request_too_large" + else CAPACITY_CONFIGURATION_HINT + ) + actionable_reason = f"{reason} {hint}" turn_metadata["large_context_capacity_block_reason"] = actionable_reason - raise LargeContextCapacityError(actionable_reason) + raise LargeContextCapacityError( + actionable_reason, + status=assessment.status, + ) + + +def require_current_selector_capacity( + selector: Any, + turn_metadata: dict[str, Any], + *, + reason: str, +) -> None: + """Revalidate the selector's bound physical head without executing it.""" + + if turn_metadata.get("large_context_capacity_blocked") is True: + from opensquilla.engine.capacity_admission import LargeContextCapacityError + + raise LargeContextCapacityError( + str(turn_metadata.get("large_context_capacity_block_reason") or reason), + status=turn_metadata.get("large_context_capacity_status"), + ) + _require_provider_config_capacity( + getattr(selector, "current_config", None), + turn_metadata, + reason=reason, + ) def _materialize_fallback_configs( @@ -560,7 +677,10 @@ def apply_model_override( turn_metadata.get("large_context_capacity_block_reason") or "No deployment has proven capacity for this attachment request." ) - raise LargeContextCapacityError(reason) + raise LargeContextCapacityError( + reason, + status=turn_metadata.get("large_context_capacity_status"), + ) if tier_provider_config is not None and hasattr(selector, "override_provider_config"): _require_provider_config_capacity( diff --git a/src/opensquilla/engine/steps/squilla_router.py b/src/opensquilla/engine/steps/squilla_router.py index 96dfeae20..a86f37fbb 100644 --- a/src/opensquilla/engine/steps/squilla_router.py +++ b/src/opensquilla/engine/steps/squilla_router.py @@ -12,6 +12,7 @@ import os import threading import time +from dataclasses import dataclass from inspect import Parameter, signature from pathlib import Path from typing import Any, Protocol, cast @@ -19,8 +20,10 @@ import structlog from opensquilla.engine.capacity_admission import ( + CAPACITY_REDUCTION_HINT, MAX_THINKING_BUDGET_TOKENS, - model_has_request_capacity, + ModelRequestCapacityAssessment, + assess_model_request_capacity, ) from opensquilla.engine.pipeline import TurnContext from opensquilla.engine.pricing import lookup_price @@ -1153,7 +1156,15 @@ def _block_large_context_route( return ctx -def _capacity_safe_tier( +@dataclass(frozen=True, slots=True) +class _TierCapacityAssessment: + name: str + provider: str + model: str + assessment: ModelRequestCapacityAssessment + + +def _capacity_tier_assessments( ctx: TurnContext, router_cfg: object, tiers: dict, @@ -1166,8 +1177,8 @@ def _capacity_safe_tier( active_provider_only: bool = False, thinking_mode: str | None = None, rollout_phase: str = "full", -) -> str | None: - """Pick the first candidate whose deployment definitely fits this turn.""" +) -> list[_TierCapacityAssessment]: + """Assess eligible candidates without collapsing unknown and too-large.""" minimum_index = tier_index(minimum_tier) active_provider = str( @@ -1177,6 +1188,7 @@ def _capacity_safe_tier( _llm_capacity_overrides(ctx) ) ordered = list(dict.fromkeys(candidate_names)) + assessments: list[_TierCapacityAssessment] = [] for name in ordered: candidate_index = tier_index(name) if minimum_index >= 0: @@ -1206,24 +1218,69 @@ def _capacity_safe_tier( thinking_mode=thinking_mode, rollout_phase=rollout_phase, ) - if model_has_request_capacity( - provider=provider, - model=tier.model, - material_tokens=material_tokens, - request_input_tokens=request_input_tokens, - thinking_budget_tokens=thinking_budget, - context_window_override_tokens=context_window_override, - max_output_override_tokens=max_output_override, - provider_request_proof_max_chars=proof_max_chars, - api_key=api_key, - base_url=base_url, - proxy=proxy, - ): - return name + assessments.append( + _TierCapacityAssessment( + name=name, + provider=provider, + model=tier.model, + assessment=assess_model_request_capacity( + provider=provider, + model=tier.model, + material_tokens=material_tokens, + request_input_tokens=request_input_tokens, + thinking_budget_tokens=thinking_budget, + context_window_override_tokens=context_window_override, + max_output_override_tokens=max_output_override, + provider_request_proof_max_chars=proof_max_chars, + api_key=api_key, + base_url=base_url, + proxy=proxy, + ), + ) + ) + return assessments + + +def _capacity_safe_tier( + ctx: TurnContext, + router_cfg: object, + tiers: dict, + candidate_names: list[str], + *, + minimum_tier: str | None, + material_tokens: int, + request_input_tokens: int = 0, + requires_image: bool = False, + active_provider_only: bool = False, + thinking_mode: str | None = None, + rollout_phase: str = "full", +) -> str | None: + """Pick the first candidate whose deployment definitely fits this turn.""" + + for candidate in _capacity_tier_assessments( + ctx, + router_cfg, + tiers, + candidate_names, + minimum_tier=minimum_tier, + material_tokens=material_tokens, + request_input_tokens=request_input_tokens, + requires_image=requires_image, + active_provider_only=active_provider_only, + thinking_mode=thinking_mode, + rollout_phase=rollout_phase, + ): + if candidate.assessment.fits: + return candidate.name return None -async def finalize_squilla_router_capacity(ctx: TurnContext) -> TurnContext: +async def finalize_squilla_router_capacity( + ctx: TurnContext, + *, + allow_compaction_retry: bool = False, + retry_after_compaction: bool = False, +) -> TurnContext: """Revalidate attachment routes against the complete pipeline request. This runs after prompt/tool/skill shaping but before selector binding. It @@ -1241,7 +1298,26 @@ async def finalize_squilla_router_capacity(ctx: TurnContext) -> TurnContext: return ctx if ctx.metadata.get("large_context_capacity_blocked") is True: return ctx + retry_pending = ( + ctx.metadata.get("large_context_capacity_retry_pending") is True + ) + if retry_pending and not retry_after_compaction: + # Selector binding and Agent bootstrap may revisit the finalized turn, + # but only the runtime's post-compaction boundary may consume the retry. + return ctx + if retry_after_compaction: + if ( + not retry_pending + or ctx.metadata.get("large_context_capacity_retry_attempted") is True + ): + return _block_large_context_route( + ctx, + "Attachment capacity admission retry state was invalid.", + include_configuration_hint=False, + ) + ctx.metadata["large_context_capacity_retry_attempted"] = True if ctx.metadata.get("routing_history_capacity_estimate_complete") is False: + ctx.metadata["large_context_capacity_retry_pending"] = False return _block_large_context_route( ctx, "Attachment request capacity could not be proven because session " @@ -1287,7 +1363,16 @@ async def finalize_squilla_router_capacity(ctx: TurnContext) -> TurnContext: dict.fromkeys(valid_tiers), key=lambda name: (0, tier_index(name)) if tier_index(name) >= 0 else (1, 0), ) - candidates = ([selected_tier] if selected_tier else []) + valid_tiers + provisional_tier = str( + ctx.metadata.get("large_context_capacity_provisional_tier") or "" + ).strip() + candidates = ( + [provisional_tier] + if retry_after_compaction and provisional_tier in tiers + else [] + if retry_after_compaction + else ([selected_tier] if selected_tier else []) + valid_tiers + ) admission_minimum_tier = minimum_tier selected_index = tier_index(selected_tier) minimum_index = tier_index(minimum_tier) @@ -1295,8 +1380,10 @@ async def finalize_squilla_router_capacity(ctx: TurnContext) -> TurnContext: # Capacity revalidation may keep or upgrade a semantic route. It must # never turn a c2/c3 decision into a cheaper lower-complexity route. admission_minimum_tier = selected_tier + if retry_after_compaction: + admission_minimum_tier = provisional_tier or selected_tier thinking_mode = ctx.metadata.get("thinking_mode") - capacity_tier = _capacity_safe_tier( + assessments = _capacity_tier_assessments( ctx, router_cfg, tiers, @@ -1309,13 +1396,118 @@ async def finalize_squilla_router_capacity(ctx: TurnContext) -> TurnContext: thinking_mode=(thinking_mode if isinstance(thinking_mode, str) else None), rollout_phase=str(ctx.metadata.get("rollout_phase") or "full"), ) + fitting = next( + (candidate for candidate in assessments if candidate.assessment.fits), + None, + ) + capacity_tier = fitting.name if fitting is not None else None + if capacity_tier is None and not retry_after_compaction: + history_tokens = max( + 0, + _token_estimate(ctx.metadata.get("large_context_history_tokens")) or 0, + ) + request_without_history = max(0, request_input_tokens - history_tokens) + configured_compaction = getattr(ctx.config, "compaction", None) + compaction_enabled = bool( + configured_compaction is None + or getattr(configured_compaction, "enabled", True) + ) + plausible_known = [ + candidate + for candidate in assessments + if candidate.assessment.status + == "known_capacity_request_too_large" + and candidate.assessment.safe_input_tokens is not None + and history_tokens > 0 + and request_without_history + <= candidate.assessment.safe_input_tokens + and not tier_ensemble_active(tiers, candidate.name) + ] + if ( + allow_compaction_retry + and compaction_enabled + and plausible_known + and not bool( + getattr(getattr(ctx.config, "llm_ensemble", None), "enabled", False) + ) + and not ctx.session_key.startswith(("cron:", "subagent:")) + ): + provisional = max( + plausible_known, + key=lambda candidate: candidate.assessment.safe_input_tokens or 0, + ) + capacity_tier = provisional.name + ctx.metadata["large_context_capacity_status"] = ( + "known_capacity_request_too_large" + ) + ctx.metadata["large_context_capacity_retry_pending"] = True + ctx.metadata.pop("large_context_capacity_retry_attempted", None) + ctx.metadata["large_context_capacity_provisional_tier"] = provisional.name + ctx.metadata["large_context_capacity_provisional_provider"] = ( + provisional.provider + ) + ctx.metadata["large_context_capacity_provisional_model"] = provisional.model + ctx.metadata["large_context_capacity_provisional_safe_input_tokens"] = ( + provisional.assessment.safe_input_tokens + ) + ctx.metadata["large_context_capacity_retry_history_tokens"] = history_tokens + ctx.metadata["large_context_capacity_retry_request_without_history_tokens"] = ( + request_without_history + ) + else: + has_unknown = not assessments or any( + candidate.assessment.status == "capacity_unknown" + for candidate in assessments + ) + ctx.metadata["large_context_capacity_status"] = ( + "capacity_unknown" + if has_unknown + else "known_capacity_request_too_large" + ) + if has_unknown: + return _block_large_context_route( + ctx, + "No SquillaRouter deployment has proven capacity for the complete " + "attachment request.", + ) + return _block_large_context_route( + ctx, + "The complete attachment request exceeds every eligible " + f"deployment's known safe input capacity. {CAPACITY_REDUCTION_HINT}", + include_configuration_hint=False, + ) if capacity_tier is None: + ctx.metadata["large_context_capacity_retry_pending"] = False + has_unknown = not assessments or any( + candidate.assessment.status == "capacity_unknown" + for candidate in assessments + ) + ctx.metadata["large_context_capacity_status"] = ( + "capacity_unknown" + if has_unknown + else "known_capacity_request_too_large" + ) + if has_unknown: + return _block_large_context_route( + ctx, + "The selected attachment deployment's capacity could not be " + "proven after automatic compaction.", + ) return _block_large_context_route( ctx, - "No SquillaRouter deployment has proven capacity for the complete " - "attachment request.", + "The attachment request still exceeds the selected deployment's " + "known safe input capacity after the bounded automatic compaction " + f"attempt. {CAPACITY_REDUCTION_HINT}", + include_configuration_hint=False, ) + ctx.metadata["large_context_capacity_status"] = ( + "fits" if fitting is not None else "known_capacity_request_too_large" + ) + if retry_after_compaction: + ctx.metadata["large_context_capacity_retry_pending"] = False + ctx.metadata["large_context_capacity_retry_succeeded"] = True + tier_cfg = tiers[capacity_tier] prior_tier = selected_tier if prior_tier != capacity_tier: diff --git a/src/opensquilla/session/terminal_reply.py b/src/opensquilla/session/terminal_reply.py index edd444d8b..7a8e4eb7b 100644 --- a/src/opensquilla/session/terminal_reply.py +++ b/src/opensquilla/session/terminal_reply.py @@ -179,6 +179,23 @@ def build_terminal_reply( "preserved the recoverable state; retry with a narrower request " "or a larger-context model." ) + if error_class == "attachment_capacity_too_large": + return ( + "The attachment request still exceeds the selected deployment's " + "known context capacity. Reduce the attachment or session context, " + "run /compact, or start a new session before retrying." + ) + if error_class == "attachment_capacity_unknown": + return ( + "OpenSquilla could not verify the selected attachment deployment's " + "context capacity. For a custom or catalog-unknown model, set " + "llm.context_window_tokens to the deployment's verified context limit." + ) + if error_class == "attachment_capacity_unavailable": + return ( + "No model deployment has proven capacity for this attachment request. " + "Check the model context limit or reduce the request before retrying." + ) if ( error_class == "empty_response" and error_message == _REASONING_ONLY_OUTPUT_BUDGET_ERROR_MESSAGE diff --git a/tests/functional/test_gateway_attachment_history_e2e.py b/tests/functional/test_gateway_attachment_history_e2e.py index b925a64ba..578cbfd30 100644 --- a/tests/functional/test_gateway_attachment_history_e2e.py +++ b/tests/functional/test_gateway_attachment_history_e2e.py @@ -272,6 +272,23 @@ async def _upload_png(app: Any) -> str: return file_uuid +async def _upload_text(app: Any) -> str: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url="http://testserver", + ) as client: + response = await client.post( + "/api/v1/files/upload", + files={"file": ("capacity.txt", b"capacity material", "text/plain")}, + ) + assert response.status_code == 200, response.text + payload = response.json() + file_uuid = payload.get("file_uuid") + assert isinstance(file_uuid, str) and file_uuid.startswith("u-") + return file_uuid + + async def _send_session_turn( *, ctx: RpcContext, @@ -361,6 +378,16 @@ def _file_uuid_attachment(file_uuid: str) -> dict[str, str]: return {"file_uuid": file_uuid, "mime": "image/png", "name": "first.png"} +def _assert_persisted_png_attachment(entry: Any) -> None: + persisted = json.loads(entry.content) + attachments = persisted.get("attachments") + assert isinstance(attachments, list) and len(attachments) == 1 + attachment = attachments[0] + assert attachment["mime"] == "image/png" + assert attachment["name"] == "first.png" + assert attachment["sha256_ref"] == hashlib.sha256(_PNG_BYTES).hexdigest() + + def _deterministic_png_payload(*, seed: str, size: int = 80_000) -> bytes: """Return stable high-entropy PNG-like bytes for capacity regression fixtures.""" @@ -807,6 +834,659 @@ async def _record_router_capacity( assert persisted.compaction_count == 0 +@pytest.mark.asyncio +async def test_gateway_known_history_pressure_compacts_once_then_readmits( + _e2e_stack: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A known image deployment may bind only long enough to compact and retry.""" + + manager: SessionManager = _e2e_stack["manager"] + runner: TurnRunner = _e2e_stack["runner"] + subscription_manager: SubscriptionManager = _e2e_stack[ + "subscription_manager" + ] + sink: _EventSink = _e2e_stack["sink"] + text_provider: _RecordingProvider = _e2e_stack["text_provider"] + gate_provider: _RecordingProvider = _e2e_stack["gate_provider"] + vision_provider: _RecordingProvider = _e2e_stack["vision_provider"] + key = "agent:main:attachment-capacity-compaction-retry" + await manager.create(session_key=key, agent_id="main") + subscription_manager.subscribe_messages(sink.conn_id, key) + + # Keep the durable rows small; the route-capacity port below supplies the + # deterministic before/after estimates so this orchestration regression + # does not spend minutes tokenizing a synthetic megabyte-scale transcript. + await manager.append_message(key, "user", "historical user") + await manager.append_message( + key, + "assistant", + "historical assistant", + ) + + from opensquilla.provider.model_catalog import ModelCatalog + + catalog = ModelCatalog() + catalog.set_user_overrides( + { + f"{_PROVIDER_ID}/{_VISION_MODEL}": { + "context_window": 128_000, + "max_output_tokens": 1_024, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + + order: list[str] = [] + capacity_results: list[dict[str, Any]] = [] + capacity_observations: list[tuple[int, bool, int]] = [] + + async def _record_router_capacity( + _session_key: str, + request: Any, + **kwargs: Any, + ) -> dict[str, Any]: + assert kwargs["max_history_turns"] == 1 + assert kwargs["preserve_image_attachments"] is True + snapshot = request.transcript_snapshot + assert snapshot is not None + entries = list(await snapshot.get_entries()) + has_historical_entries = any( + str(getattr(entry, "content", "")).startswith("historical") + for entry in entries + ) + capacity_observations.append( + (snapshot.generation, has_historical_entries, len(entries)) + ) + result = { + "history_capacity_estimated_tokens": ( + 200_000 if has_historical_entries else 0 + ), + "history_capacity_message_count": 2 if has_historical_entries else 0, + "history_capacity_estimate_complete": True, + } + order.append("capacity") + capacity_results.append(dict(result)) + return result + + monkeypatch.setattr( + runner, + "_router_history_capacity_for_request", + _record_router_capacity, + ) + preflight_calls = 0 + snapshot_generations: list[tuple[int, int]] = [] + run_preflight = runner._maybe_preflight_compact + + async def _record_real_preflight( + *args: Any, + **kwargs: Any, + ) -> None: + nonlocal preflight_calls + preflight_calls += 1 + snapshot = kwargs.get("transcript_snapshot") + assert snapshot is not None + generation_before = snapshot.generation + await run_preflight(*args, **kwargs) + snapshot_generations.append((generation_before, snapshot.generation)) + + monkeypatch.setattr( + runner, + "_maybe_preflight_compact", + _record_real_preflight, + ) + + from opensquilla.session import compaction as compaction_module + from opensquilla.session.compaction import CompactionResult + + estimate_replay_tokens = compaction_module.estimate_entry_model_replay_tokens + + def _force_historical_token_pressure(entry: Any) -> int: + if str(getattr(entry, "content", "")).startswith("historical"): + return 100_000 + return estimate_replay_tokens(entry) + + monkeypatch.setattr( + compaction_module, + "estimate_entry_model_replay_tokens", + _force_historical_token_pressure, + ) + checkpoint_calls = 0 + + async def _record_safe_checkpoint(*_args: Any, **_kwargs: Any) -> bool: + nonlocal checkpoint_calls + checkpoint_calls += 1 + return True + + monkeypatch.setattr( + runner, + "_record_checkpoint_before_compaction", + _record_safe_checkpoint, + ) + compact_calls = 0 + + async def _install_compacted_history( + session_key: str, + _context_window_tokens: int, + _config: Any, + **kwargs: Any, + ) -> CompactionResult: + nonlocal compact_calls + compact_calls += 1 + order.append("compact") + truncated = await manager.truncate(session_key, max_messages=1) + assert truncated["truncated"] is True + return CompactionResult( + summary="Compacted historical context.", + kept_entries=[], + removed_count=2, + chunks_processed=1, + ) + + monkeypatch.setattr( + manager, + "compact_with_result", + _install_compacted_history, + ) + original_vision_chat = vision_provider.chat + + async def _guarded_vision_chat( + messages: list[Message], + tools: list[Any] | None = None, + config: ChatConfig | None = None, + ) -> AsyncIterator[Any]: + order.append("provider") + assert preflight_calls == 1 + assert order.count("capacity") == 2 + async for event in original_vision_chat(messages, tools, config): + yield event + + monkeypatch.setattr(vision_provider, "chat", _guarded_vision_chat) + + captured_turns: list[Any] = [] + original_bootstrap = runner._agent_bootstrap_stage.run + + async def _capture_turn(inp: Any) -> Any: + captured_turns.append(inp.turn) + return await original_bootstrap(inp) + + monkeypatch.setattr(runner._agent_bootstrap_stage, "run", _capture_turn) + + file_uuid = await _upload_png(_e2e_stack["app"]) + await _send_session_turn( + ctx=_e2e_stack["ctx"], + key=key, + sink=sink, + message="Describe only the current image after compacting old history.", + attachments=[_file_uuid_attachment(file_uuid)], + ) + + assert order == ["capacity", "compact", "capacity", "provider"] + assert preflight_calls == 1 + assert checkpoint_calls == 1 + assert compact_calls == 1 + assert snapshot_generations == [(0, 1)] + assert capacity_observations[0][:2] == (0, True) + assert capacity_observations[1][:2] == (1, False) + assert capacity_observations[0][2] > capacity_observations[1][2] + assert len(capacity_results) == 2 + assert ( + capacity_results[0]["history_capacity_estimated_tokens"] + > capacity_results[1]["history_capacity_estimated_tokens"] + ) + assert len(text_provider.calls) == 0 + assert len(gate_provider.calls) == 0 + assert len(vision_provider.calls) == 1 + assert captured_turns + turn_metadata = captured_turns[-1].metadata + assert turn_metadata["large_context_capacity_retry_attempted"] is True + assert turn_metadata["large_context_capacity_retry_succeeded"] is True + assert turn_metadata["large_context_capacity_retry_pending"] is False + assert turn_metadata["large_context_capacity_status"] == "fits" + assert turn_metadata["large_context_capacity_provisional_model"] == _VISION_MODEL + assert turn_metadata["large_context_capacity_compaction_preflight_invoked"] is True + assert not _event_payloads(sink, "session.event.error") + sent_messages = vision_provider.calls[0]["messages"] + assert _message_has_image(sent_messages[-1]) + sent_images = _message_image_blocks(sent_messages[-1]) + assert len(sent_images) == 1 + assert base64.b64decode(sent_images[0].data) == _PNG_BYTES + persisted_transcript = await manager.get_transcript(key) + persisted_user_rows = [ + entry for entry in persisted_transcript if entry.role == "user" + ] + assert len(persisted_user_rows) == 1 + _assert_persisted_png_attachment(persisted_user_rows[0]) + + +@pytest.mark.asyncio +async def test_gateway_post_compaction_capacity_failure_never_calls_provider( + _e2e_stack: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager: SessionManager = _e2e_stack["manager"] + runner: TurnRunner = _e2e_stack["runner"] + subscription_manager: SubscriptionManager = _e2e_stack[ + "subscription_manager" + ] + sink: _EventSink = _e2e_stack["sink"] + text_provider: _RecordingProvider = _e2e_stack["text_provider"] + gate_provider: _RecordingProvider = _e2e_stack["gate_provider"] + vision_provider: _RecordingProvider = _e2e_stack["vision_provider"] + key = "agent:main:attachment-capacity-compaction-insufficient" + await manager.create(session_key=key, agent_id="main") + subscription_manager.subscribe_messages(sink.conn_id, key) + await manager.append_message(key, "user", "historical user") + await manager.append_message(key, "assistant", "historical assistant") + + from opensquilla.provider.model_catalog import ModelCatalog + + catalog = ModelCatalog() + catalog.set_user_overrides( + { + f"{_PROVIDER_ID}/{_VISION_MODEL}": { + "context_window": 128_000, + "max_output_tokens": 1_024, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + + order: list[str] = [] + capacity_calls = 0 + capacity_observations: list[tuple[int, int, int]] = [] + + async def _still_too_large( + _session_key: str, + request: Any, + **_kwargs: Any, + ) -> dict[str, Any]: + nonlocal capacity_calls + capacity_calls += 1 + snapshot = request.transcript_snapshot + assert snapshot is not None + entries = list(await snapshot.get_entries()) + historical_entry_count = sum( + str(getattr(entry, "content", "")).startswith("historical") + for entry in entries + ) + capacity_observations.append( + (snapshot.generation, historical_entry_count, len(entries)) + ) + order.append("capacity") + return { + "history_capacity_estimated_tokens": ( + 200_000 if historical_entry_count > 1 else 150_000 + ), + "history_capacity_message_count": historical_entry_count, + "history_capacity_estimate_complete": True, + } + + monkeypatch.setattr( + runner, + "_router_history_capacity_for_request", + _still_too_large, + ) + preflight_calls = 0 + snapshot_generations: list[tuple[int, int]] = [] + run_preflight = runner._maybe_preflight_compact + + async def _record_real_preflight( + *args: Any, + **kwargs: Any, + ) -> None: + nonlocal preflight_calls + preflight_calls += 1 + snapshot = kwargs.get("transcript_snapshot") + assert snapshot is not None + generation_before = snapshot.generation + await run_preflight(*args, **kwargs) + snapshot_generations.append((generation_before, snapshot.generation)) + + monkeypatch.setattr( + runner, + "_maybe_preflight_compact", + _record_real_preflight, + ) + + from opensquilla.session import compaction as compaction_module + from opensquilla.session.compaction import CompactionResult + + estimate_replay_tokens = compaction_module.estimate_entry_model_replay_tokens + + def _force_historical_token_pressure(entry: Any) -> int: + if str(getattr(entry, "content", "")).startswith("historical"): + return 100_000 + return estimate_replay_tokens(entry) + + monkeypatch.setattr( + compaction_module, + "estimate_entry_model_replay_tokens", + _force_historical_token_pressure, + ) + checkpoint_calls = 0 + + async def _record_safe_checkpoint(*_args: Any, **_kwargs: Any) -> bool: + nonlocal checkpoint_calls + checkpoint_calls += 1 + return True + + monkeypatch.setattr( + runner, + "_record_checkpoint_before_compaction", + _record_safe_checkpoint, + ) + compact_calls = 0 + + async def _install_insufficient_compacted_history( + session_key: str, + _context_window_tokens: int, + _config: Any, + **_kwargs: Any, + ) -> CompactionResult: + nonlocal compact_calls + compact_calls += 1 + order.append("compact") + truncated = await manager.truncate(session_key, max_messages=2) + assert truncated["truncated"] is True + return CompactionResult( + summary="Compacted historical context.", + kept_entries=[], + removed_count=1, + chunks_processed=1, + ) + + monkeypatch.setattr( + manager, + "compact_with_result", + _install_insufficient_compacted_history, + ) + captured_turns: list[Any] = [] + original_bootstrap = runner._agent_bootstrap_stage.run + + async def _capture_turn(inp: Any) -> Any: + captured_turns.append(inp.turn) + return await original_bootstrap(inp) + + monkeypatch.setattr(runner._agent_bootstrap_stage, "run", _capture_turn) + + file_uuid = await _upload_png(_e2e_stack["app"]) + event_count_before = len(sink.events) + await _send_session_turn( + ctx=_e2e_stack["ctx"], + key=key, + sink=sink, + message="Describe the current image after trying to compact old history.", + attachments=[_file_uuid_attachment(file_uuid)], + expected_error_code="attachment_capacity_too_large", + ) + + assert order == ["capacity", "compact", "capacity"] + assert capacity_calls == 2 + assert preflight_calls == 1 + assert checkpoint_calls == 1 + assert compact_calls == 1 + assert snapshot_generations == [(0, 1)] + assert capacity_observations[0][:2] == (0, 2) + assert capacity_observations[1][:2] == (1, 1) + assert capacity_observations[0][2] > capacity_observations[1][2] + assert len(text_provider.calls) == 0 + assert len(gate_provider.calls) == 0 + assert len(vision_provider.calls) == 0 + errors = [ + payload + for event, payload in sink.events[event_count_before:] + if event == "session.event.error" + ] + assert errors[-1]["code"] == "attachment_capacity_too_large" + assert "/compact" in errors[-1]["message"] + assert "llm.context_window_tokens" not in errors[-1]["message"] + assert captured_turns + turn_metadata = captured_turns[-1].metadata + assert turn_metadata["large_context_capacity_retry_attempted"] is True + assert turn_metadata["large_context_capacity_retry_pending"] is False + assert turn_metadata["large_context_capacity_blocked"] is True + assert turn_metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + persisted_transcript = await manager.get_transcript(key) + persisted_user_rows = [ + entry for entry in persisted_transcript if entry.role == "user" + ] + assert len(persisted_user_rows) == 1 + _assert_persisted_png_attachment(persisted_user_rows[0]) + + +@pytest.mark.asyncio +async def test_gateway_capacity_unknown_uses_stable_code_without_provider_call( + _e2e_stack: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, +) -> None: + config: GatewayConfig = _e2e_stack["config"] + manager: SessionManager = _e2e_stack["manager"] + runner: TurnRunner = _e2e_stack["runner"] + subscription_manager: SubscriptionManager = _e2e_stack[ + "subscription_manager" + ] + sink: _EventSink = _e2e_stack["sink"] + text_provider: _RecordingProvider = _e2e_stack["text_provider"] + gate_provider: _RecordingProvider = _e2e_stack["gate_provider"] + vision_provider: _RecordingProvider = _e2e_stack["vision_provider"] + unknown_model = "catalog-unknown-vision" + config.squilla_router.tiers["image_model"]["model"] = unknown_model + config.llm.context_window_tokens = 0 + + from opensquilla.provider.model_catalog import ModelCatalog + + monkeypatch.setattr( + "opensquilla.provider.model_catalog._shared_catalog", + ModelCatalog(), + ) + runtime_catalog = runner._model_catalog + assert runtime_catalog is not None + monkeypatch.setattr( + runtime_catalog, + "get_capabilities", + lambda _model_id, **_kwargs: ModelCapabilities(supports_vision=True), + ) + monkeypatch.setattr( + runtime_catalog, + "resolve_vision_support", + lambda _model_id, **_kwargs: "supported", + ) + key = "agent:main:attachment-capacity-unknown" + await manager.create(session_key=key, agent_id="main") + subscription_manager.subscribe_messages(sink.conn_id, key) + captured_turns: list[Any] = [] + finalize_capacity = squilla_router_step.finalize_squilla_router_capacity + + async def _capture_finalized_turn(turn: Any, **kwargs: Any) -> Any: + finalized = await finalize_capacity(turn, **kwargs) + captured_turns.append(finalized) + return finalized + + monkeypatch.setattr( + "opensquilla.engine.steps.finalize_squilla_router_capacity", + _capture_finalized_turn, + ) + + file_uuid = await _upload_png(_e2e_stack["app"]) + event_count_before = len(sink.events) + await _send_session_turn( + ctx=_e2e_stack["ctx"], + key=key, + sink=sink, + message="Describe the current image.", + attachments=[_file_uuid_attachment(file_uuid)], + expected_error_code="attachment_capacity_unknown", + ) + + errors = [ + payload + for event, payload in sink.events[event_count_before:] + if event == "session.event.error" + ] + expected_reply = ( + "OpenSquilla could not verify the selected attachment deployment's context " + "capacity. For a custom or catalog-unknown model, set " + "llm.context_window_tokens to the deployment's verified context limit." + ) + assert errors[-1]["code"] == "attachment_capacity_unknown" + assert errors[-1]["message"] == expected_reply + assert errors[-1]["terminal_message"] == expected_reply + assert "internal" not in errors[-1]["message"].lower() + assert len(text_provider.calls) == 0 + assert len(gate_provider.calls) == 0 + assert len(vision_provider.calls) == 0 + assert captured_turns + turn_metadata = captured_turns[-1].metadata + assert turn_metadata["large_context_capacity_blocked"] is True + assert turn_metadata["large_context_capacity_status"] == "capacity_unknown" + assert "large_context_capacity_retry_pending" not in turn_metadata + assert "llm.context_window_tokens" in turn_metadata[ + "large_context_capacity_block_reason" + ] + persisted_transcript = await manager.get_transcript(key) + persisted_user_rows = [ + entry for entry in persisted_transcript if entry.role == "user" + ] + assert len(persisted_user_rows) == 1 + _assert_persisted_png_attachment(persisted_user_rows[0]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("ensemble_scope", ["global", "tier"]) +async def test_gateway_ensemble_capacity_block_never_reaches_provider( + _e2e_stack: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + ensemble_scope: str, +) -> None: + config: GatewayConfig = _e2e_stack["config"] + manager: SessionManager = _e2e_stack["manager"] + runner: TurnRunner = _e2e_stack["runner"] + subscription_manager: SubscriptionManager = _e2e_stack[ + "subscription_manager" + ] + sink: _EventSink = _e2e_stack["sink"] + text_provider: _RecordingProvider = _e2e_stack["text_provider"] + gate_provider: _RecordingProvider = _e2e_stack["gate_provider"] + vision_provider: _RecordingProvider = _e2e_stack["vision_provider"] + usage_sink: _UsageSink = _e2e_stack["usage_sink"] + tier_name = "c0" if ensemble_scope == "global" else "c3" + tier: dict[str, Any] = { + "provider": _PROVIDER_ID, + "model": _TEXT_MODEL, + "supports_image": False, + } + if ensemble_scope == "global": + config.llm_ensemble.enabled = True + config.llm_ensemble.selection_mode = "static_openrouter_b5" + else: + tier["ensemble_enabled"] = True + config.squilla_router.tiers = {tier_name: tier} + config.squilla_router.default_tier = tier_name + monkeypatch.setattr( + squilla_router_step, + "_get_strategy", + lambda _config: _TextTierStrategy(), + ) + + from opensquilla.provider.model_catalog import ModelCatalog + + catalog = ModelCatalog() + catalog.set_user_overrides( + { + f"{_PROVIDER_ID}/{_TEXT_MODEL}": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + capacity_calls = 0 + + async def _history_pressure( + _session_key: str, + request: Any, + **_kwargs: Any, + ) -> dict[str, Any]: + nonlocal capacity_calls + capacity_calls += 1 + assert request.transcript_snapshot is not None + return { + "history_capacity_estimated_tokens": 30_000, + "history_capacity_message_count": 2, + "history_capacity_estimate_complete": True, + } + + monkeypatch.setattr( + runner, + "_router_history_capacity_for_request", + _history_pressure, + ) + preflight_calls = 0 + run_preflight = runner._maybe_preflight_compact + + async def _record_preflight(*args: Any, **kwargs: Any) -> None: + nonlocal preflight_calls + preflight_calls += 1 + await run_preflight(*args, **kwargs) + + monkeypatch.setattr(runner, "_maybe_preflight_compact", _record_preflight) + captured_turns: list[Any] = [] + finalize_capacity = squilla_router_step.finalize_squilla_router_capacity + + async def _capture_finalized_turn(turn: Any, **kwargs: Any) -> Any: + finalized = await finalize_capacity(turn, **kwargs) + captured_turns.append(finalized) + return finalized + + monkeypatch.setattr( + "opensquilla.engine.steps.finalize_squilla_router_capacity", + _capture_finalized_turn, + ) + key = f"agent:main:attachment-capacity-{ensemble_scope}-ensemble" + await manager.create(session_key=key, agent_id="main") + subscription_manager.subscribe_messages(sink.conn_id, key) + + file_uuid = await _upload_text(_e2e_stack["app"]) + event_count_before = len(sink.events) + await _send_session_turn( + ctx=_e2e_stack["ctx"], + key=key, + sink=sink, + message="Use the current text attachment.", + attachments=[ + { + "file_uuid": file_uuid, + "mime": "text/plain", + "name": "capacity.txt", + } + ], + expected_error_code="attachment_capacity_too_large", + ) + + errors = [ + payload + for event, payload in sink.events[event_count_before:] + if event == "session.event.error" + ] + assert errors[-1]["code"] == "attachment_capacity_too_large" + assert capacity_calls == 1 + assert preflight_calls == 0 + assert len(text_provider.calls) == 0 + assert len(gate_provider.calls) == 0 + assert len(vision_provider.calls) == 0 + assert usage_sink.started == [] + assert captured_turns + turn_metadata = captured_turns[-1].metadata + assert turn_metadata["large_context_capacity_blocked"] is True + assert turn_metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + assert "large_context_capacity_retry_pending" not in turn_metadata + + @pytest.mark.asyncio async def test_historical_image_material_is_not_replayed_without_vision_support( _e2e_stack: dict[str, Any], diff --git a/tests/test_gateway/test_terminal_reply.py b/tests/test_gateway/test_terminal_reply.py index 3e5c37df5..f42f12f1a 100644 --- a/tests/test_gateway/test_terminal_reply.py +++ b/tests/test_gateway/test_terminal_reply.py @@ -144,6 +144,24 @@ }, "single-model routing mode", ), + ( + { + "status": "failed", + "terminal_reason": "error", + "error_class": "attachment_capacity_too_large", + "error_message": "internal estimator detail", + }, + "/compact", + ), + ( + { + "status": "failed", + "terminal_reason": "error", + "error_class": "attachment_capacity_unknown", + "error_message": "internal catalog detail", + }, + "llm.context_window_tokens", + ), ], ) def test_build_terminal_reply_returns_user_readable_messages( @@ -192,6 +210,44 @@ def test_ensemble_multimodal_reply_is_actionable_and_stable() -> None: ) +@pytest.mark.parametrize( + ("error_class", "expected_reply"), + [ + ( + "attachment_capacity_too_large", + "The attachment request still exceeds the selected deployment's known " + "context capacity. Reduce the attachment or session context, run /compact, " + "or start a new session before retrying.", + ), + ( + "attachment_capacity_unknown", + "OpenSquilla could not verify the selected attachment deployment's context " + "capacity. For a custom or catalog-unknown model, set " + "llm.context_window_tokens to the deployment's verified context limit.", + ), + ( + "attachment_capacity_unavailable", + "No model deployment has proven capacity for this attachment request. Check " + "the model context limit or reduce the request before retrying.", + ), + ], +) +def test_attachment_capacity_replies_are_actionable_and_stable( + error_class: str, + expected_reply: str, +) -> None: + reply = build_terminal_reply( + { + "status": "failed", + "terminal_reason": "error", + "error_class": error_class, + "error_message": "private internal capacity detail must not win", + } + ) + + assert reply == expected_reply + + def test_image_input_unsupported_reply_is_actionable_and_stable() -> None: reply = build_terminal_reply( { diff --git a/tests/test_model_router_behavior.py b/tests/test_model_router_behavior.py index 4fb0593bd..44aa012dd 100644 --- a/tests/test_model_router_behavior.py +++ b/tests/test_model_router_behavior.py @@ -692,6 +692,372 @@ async def route(history_tokens: int) -> TurnContext: ) +@pytest.mark.asyncio +async def test_known_history_pressure_gets_one_fixed_tier_compaction_retry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/history-bound": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "openrouter", + "model": "history-bound", + "thinking_level": "off", + } + } + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 30_000 + ctx.metadata["routing_history_capacity_message_count"] = 12 + + provisional = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert provisional.metadata["large_context_capacity_retry_pending"] is True + assert provisional.metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + assert provisional.metadata["large_context_capacity_provisional_tier"] == "c0" + assert provisional.metadata["large_context_capacity_provisional_model"] == ( + "history-bound" + ) + assert "large_context_capacity_blocked" not in provisional.metadata + + provisional.metadata["routing_history_capacity_estimated_tokens"] = 1_000 + provisional.metadata["routing_history_capacity_message_count"] = 2 + admitted = await finalize_squilla_router_capacity( + provisional, + retry_after_compaction=True, + ) + + assert admitted.metadata["large_context_capacity_retry_attempted"] is True + assert admitted.metadata["large_context_capacity_retry_pending"] is False + assert admitted.metadata["large_context_capacity_retry_succeeded"] is True + assert admitted.metadata["large_context_capacity_status"] == "fits" + assert admitted.metadata["routed_tier"] == "c0" + assert admitted.model == "history-bound" + + +@pytest.mark.asyncio +async def test_multi_candidate_retry_rechecks_only_largest_provisional_tier( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/history-tight": { + "context_window": 32_000, + "max_output_tokens": 4_000, + }, + "openrouter/history-wide": { + "context_window": 64_000, + "max_output_tokens": 4_000, + }, + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + assessed_models: list[str] = [] + assess_capacity = squilla_router_step.assess_model_request_capacity + + def _record_assessed_model(**kwargs: object): + assessed_models.append(str(kwargs.get("model") or "")) + return assess_capacity(**kwargs) + + monkeypatch.setattr( + squilla_router_step, + "assess_model_request_capacity", + _record_assessed_model, + ) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "openrouter", + "model": "history-tight", + "thinking_level": "off", + }, + "c1": { + "provider": "openrouter", + "model": "history-wide", + "thinking_level": "off", + }, + } + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 70_000 + ctx.metadata["routing_history_capacity_message_count"] = 20 + + provisional = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert provisional.metadata["large_context_capacity_retry_pending"] is True + assert provisional.metadata["large_context_capacity_provisional_tier"] == "c1" + assert provisional.metadata["large_context_capacity_provisional_provider"] == ( + "openrouter" + ) + assert provisional.metadata["large_context_capacity_provisional_model"] == ( + "history-wide" + ) + + provisional.metadata["routing_history_capacity_estimated_tokens"] = 1_000 + provisional.metadata["routing_history_capacity_message_count"] = 2 + assessed_models.clear() + admitted = await finalize_squilla_router_capacity( + provisional, + retry_after_compaction=True, + ) + + assert assessed_models == ["history-wide"] + assert admitted.metadata["large_context_capacity_retry_succeeded"] is True + assert admitted.metadata["routed_tier"] == "c1" + assert admitted.model == "history-wide" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("ensemble_scope", ["global", "tier"]) +async def test_ensemble_route_never_schedules_attachment_compaction_retry( + monkeypatch: pytest.MonkeyPatch, + ensemble_scope: str, +) -> None: + tier_name = "c0" if ensemble_scope == "global" else "c3" + fake_strategy(monkeypatch, tier_name, 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/history-bound": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + tier = { + "provider": "openrouter", + "model": "history-bound", + "thinking_level": "off", + } + if ensemble_scope == "global": + ctx.config.llm_ensemble.enabled = True + else: + tier["ensemble_enabled"] = True + ctx.config.squilla_router.tiers = {tier_name: tier} + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 30_000 + ctx.metadata["routing_history_capacity_message_count"] = 12 + + blocked = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert blocked.metadata["large_context_capacity_blocked"] is True + assert blocked.metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + assert "large_context_capacity_retry_pending" not in blocked.metadata + + +@pytest.mark.asyncio +async def test_post_compaction_retry_still_too_large_fails_actionably( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/history-bound": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "openrouter", + "model": "history-bound", + "thinking_level": "off", + } + } + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 30_000 + + provisional = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + assert provisional.metadata["large_context_capacity_retry_pending"] is True + + provisional.metadata["routing_history_capacity_estimated_tokens"] = 25_000 + blocked = await finalize_squilla_router_capacity( + provisional, + retry_after_compaction=True, + ) + + assert blocked.metadata["large_context_capacity_retry_attempted"] is True + assert blocked.metadata["large_context_capacity_retry_pending"] is False + assert blocked.metadata["large_context_capacity_blocked"] is True + assert blocked.metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + assert "/compact" in blocked.metadata["large_context_capacity_block_reason"] + assert "llm.context_window_tokens" not in blocked.metadata[ + "large_context_capacity_block_reason" + ] + + +@pytest.mark.asyncio +async def test_disabled_compaction_does_not_get_provisional_binding( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/history-bound": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.compaction.enabled = False + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "openrouter", + "model": "history-bound", + "thinking_level": "off", + } + } + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 30_000 + + blocked = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert blocked.metadata["large_context_capacity_blocked"] is True + assert "large_context_capacity_retry_pending" not in blocked.metadata + assert blocked.metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + + +@pytest.mark.asyncio +async def test_known_fixed_envelope_does_not_schedule_compaction_retry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openrouter/fixed-too-large": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "openrouter", + "model": "fixed-too-large", + "thinking_level": "off", + } + } + ctx.system_prompt = "s" * 100_000 + ctx.metadata["attachment_material_estimated_tokens"] = 8_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 1_000 + + routed = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert routed.metadata["large_context_capacity_blocked"] is True + assert routed.metadata["large_context_capacity_status"] == ( + "known_capacity_request_too_large" + ) + assert "large_context_capacity_retry_pending" not in routed.metadata + assert "/compact" in routed.metadata["large_context_capacity_block_reason"] + assert "llm.context_window_tokens" not in routed.metadata[ + "large_context_capacity_block_reason" + ] + + +@pytest.mark.asyncio +async def test_capacity_unknown_never_gets_compaction_retry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_strategy(monkeypatch, "c0", 0.91, {"route_class": "R0"}) + monkeypatch.setattr( + "opensquilla.provider.model_catalog._shared_catalog", + ModelCatalog(), + ) + ctx = make_context( + "Use the supplied material.", + attachments=[{"type": "text/plain"}], + ) + ctx.config.llm.provider = "custom" + ctx.config.llm.context_window_tokens = 0 + ctx.config.squilla_router.tiers = { + "c0": { + "provider": "custom", + "model": "unknown-history-model", + "thinking_level": "off", + } + } + ctx.metadata["attachment_material_estimated_tokens"] = 4_000 + ctx.metadata["routing_history_capacity_estimated_tokens"] = 30_000 + + routed = await finalize_squilla_router_capacity( + await apply_squilla_router(ctx), + allow_compaction_retry=True, + ) + + assert routed.metadata["large_context_capacity_blocked"] is True + assert routed.metadata["large_context_capacity_status"] == "capacity_unknown" + assert "large_context_capacity_retry_pending" not in routed.metadata + assert "llm.context_window_tokens" in routed.metadata[ + "large_context_capacity_block_reason" + ] + + @pytest.mark.asyncio async def test_catalog_unknown_custom_attachment_config_has_actionable_error( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_router_tier_contract.py b/tests/test_router_tier_contract.py index 61a81d3c1..dc7d70012 100644 --- a/tests/test_router_tier_contract.py +++ b/tests/test_router_tier_contract.py @@ -13,10 +13,14 @@ from opensquilla.context_budget import CHARS_PER_TOKEN, ContextBudgetGovernor from opensquilla.engine.capacity_admission import ( LargeContextCapacityError, + assess_model_request_capacity, model_has_request_capacity, ) from opensquilla.engine.routing import RoutingDecision -from opensquilla.engine.selector_override import apply_model_override +from opensquilla.engine.selector_override import ( + apply_model_override, + require_current_selector_capacity, +) from opensquilla.engine.steps.squilla_router import ( _apply_provider_mismatch_veto, _flag_tier_provider_mismatch, @@ -865,6 +869,109 @@ def test_complete_request_capacity_boundary_and_unknown_model_fail_closed( ) +def test_capacity_assessment_distinguishes_known_too_large_from_unknown( + monkeypatch, +) -> None: + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openai/known-model": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + + known = assess_model_request_capacity( + provider="openai", + model="known-model", + material_tokens=1, + request_input_tokens=100_000, + thinking_budget_tokens=0, + ) + unknown = assess_model_request_capacity( + provider="openai", + model="unknown-model", + material_tokens=1, + request_input_tokens=100_000, + thinking_budget_tokens=0, + ) + + assert known.status == "known_capacity_request_too_large" + assert known.safe_input_tokens is not None + assert known.required_input_tokens == 100_000 + assert unknown.status == "capacity_unknown" + assert unknown.safe_input_tokens is None + + +def test_selector_allows_only_pre_retry_binding_for_exact_known_deployment( + monkeypatch, +) -> None: + catalog = ModelCatalog() + catalog.set_user_overrides( + { + "openai/history-bound": { + "context_window": 32_000, + "max_output_tokens": 4_000, + } + } + ) + monkeypatch.setattr("opensquilla.provider.model_catalog._shared_catalog", catalog) + safe_input_tokens = ( + ContextBudgetGovernor.from_values( + context_window_tokens=32_000, + max_output_tokens=4_000, + thinking_budget_tokens=0, + context_overflow_threshold=0.85, + ).snapshot().provider_request_max_chars + // CHARS_PER_TOKEN + ) + selector = ModelSelector( + SelectorConfig( + primary=ProviderConfig( + "openai", + "history-bound", + api_key="test-key", + ) + ) + ) + metadata = { + "routing_applied": True, + "large_context_capacity_required": True, + "large_context_material_tokens": 1_000, + "large_context_request_input_tokens": safe_input_tokens + 1_000, + "large_context_history_tokens": 2_000, + "large_context_thinking_budget_tokens": 0, + "large_context_capacity_retry_pending": True, + "large_context_capacity_provisional_provider": "openai", + "large_context_capacity_provisional_model": "history-bound", + "routed_model": "history-bound", + } + + apply_model_override( + selector, + "history-bound", + turn_metadata=metadata, + realign_routed_model=False, + ) + + assert metadata["large_context_capacity_provisional_bound"] is True + assert selector.current_config.model == "history-bound" + + metadata["large_context_capacity_retry_attempted"] = True + metadata["large_context_capacity_retry_pending"] = False + with pytest.raises(LargeContextCapacityError, match="/compact"): + require_current_selector_capacity( + selector, + metadata, + reason="The post-compaction deployment is still too large.", + ) + assert "llm.context_window_tokens" not in metadata[ + "large_context_capacity_block_reason" + ] + + def test_complete_attachment_request_filters_every_fallback_without_large_floor( monkeypatch, ) -> None: