refactor(compression): consolidate fast-lane certification onto one predicate

resolve_compression_fast_lane and _compression_config_claims_fast_lane
each hand-parsed the same four config fields (provider, model,
reasoning_effort, max_output_tokens) with copy-pasted normalization and
int-coercion. Extract _fast_lane_config_fields() as the single source of
truth for both.

This also fixes a real inconsistency the duplication hid: certification
checked the literal string 'none' while _get_task_extra_body routes
reasoning_effort through parse_reasoning_effort, which treats 'false',
'disabled', and YAML boolean false as disabled too. A user writing
reasoning_effort: false got reasoning disabled but silently lost the
fast-lane cap. Certification now delegates to parse_reasoning_effort so
the two predicates can never disagree.

Regression test: every disabled-spelling certifies; empty/real efforts
do not. Mutation-checked (reverting to the literal check fails the new
test).
This commit is contained in:
kshitijk4poor
2026-08-28 12:48:40 +05:30
parent d3ccc09dc2
commit d20ca3bc80
2 changed files with 62 additions and 21 deletions
+41 -21
View File
@@ -8457,6 +8457,38 @@ class CompressionFastLane(NamedTuple):
reasoning_config: Optional[Dict[str, Any]]
def _fast_lane_config_fields(
config: Dict[str, Any],
) -> tuple[str, str, bool, Optional[int]]:
"""Extract the fast-lane certification fields from one task config.
Returns ``(provider, model, non_reasoning, cap)``:
- ``provider``/``model``: normalized (stripped; provider lowercased).
- ``non_reasoning``: True only when ``reasoning_effort`` EXPLICITLY
disables thinking. Delegates to ``parse_reasoning_effort`` so every
spelling users can write in config.yaml (``none``, ``false``,
``disabled``, YAML boolean ``false``) certifies identically —
``_get_task_extra_body`` already uses the same parser to disable
reasoning, and the two predicates must not disagree. Empty/unset
(provider default) is NOT non-reasoning.
- ``cap``: positive int from ``max_output_tokens``, else None.
Booleans are config drift, never a cap (``int(True) == 1``).
"""
from hermes_constants import parse_reasoning_effort
provider = str(config.get("provider") or "").strip().lower()
model = str(config.get("model") or "").strip()
parsed_effort = parse_reasoning_effort(config.get("reasoning_effort"))
non_reasoning = parsed_effort is not None and parsed_effort.get("enabled") is False
raw_cap = config.get("max_output_tokens")
try:
cap = 0 if isinstance(raw_cap, bool) else int(raw_cap or 0)
except (TypeError, ValueError):
cap = 0
return provider, model, non_reasoning, (cap if cap > 0 else None)
def resolve_compression_fast_lane(
actual_provider: str,
actual_model: Optional[str],
@@ -8477,44 +8509,32 @@ def resolve_compression_fast_lane(
if route_config is not None
else _get_auxiliary_task_config("compression")
)
provider = str(requested_provider or config.get("provider") or "").strip().lower()
model = str(requested_model or config.get("model") or "").strip()
effort = str(config.get("reasoning_effort") or "").strip().lower()
cfg_provider, cfg_model, non_reasoning, cap = _fast_lane_config_fields(config)
provider = str(requested_provider or "").strip().lower() or cfg_provider
model = str(requested_model or "").strip() or cfg_model
explicit_route = provider not in {"", "auto"} and model.lower() not in {"", "auto"}
provider_matches = _normalize_aux_provider(
_fallback_provider_from_label(str(actual_provider or ""))
) == _normalize_aux_provider(provider)
model_matches = str(actual_model or "").strip().lower() == model.lower()
certified = explicit_route and provider_matches and model_matches and effort == "none"
certified = explicit_route and provider_matches and model_matches and non_reasoning
if not certified:
return CompressionFastLane(False, None, None)
raw_cap = config.get("max_output_tokens")
try:
cap = 0 if isinstance(raw_cap, bool) else int(raw_cap)
except (TypeError, ValueError):
cap = 0
return CompressionFastLane(
True,
cap if cap > 0 else None,
cap,
{"enabled": False, "effort": "none"},
)
def _compression_config_claims_fast_lane(config: Dict[str, Any]) -> bool:
"""Whether task config declares fast-only controls that cannot leak."""
provider = str(config.get("provider") or "").strip().lower()
model = str(config.get("model") or "").strip().lower()
effort = str(config.get("reasoning_effort") or "").strip().lower()
raw_cap = config.get("max_output_tokens")
try:
cap = 0 if isinstance(raw_cap, bool) else int(raw_cap or 0)
except (TypeError, ValueError):
cap = 0
provider, model, non_reasoning, cap = _fast_lane_config_fields(config)
return (
provider not in {"", "auto"}
and model not in {"", "auto"}
and effort == "none"
and cap > 0
and model.lower() not in {"", "auto"}
and non_reasoning
and cap is not None
)
+21
View File
@@ -378,3 +378,24 @@ def test_fallback_cap_requires_independent_route_certification():
"enabled": False,
"effort": "none",
}
def test_reasoning_effort_aliases_certify_like_none():
"""Every spelling parse_reasoning_effort treats as disabled must certify.
_get_task_extra_body uses parse_reasoning_effort to disable reasoning for
"false"/"disabled"/YAML False exactly like "none"; the certification
predicate must agree or those users silently lose the fast lane.
"""
base = {"provider": "ollama", "model": "qwen3:8b", "max_output_tokens": 1400}
for alias in ("none", "false", "disabled", False):
lane = _resolve({**base, "reasoning_effort": alias})
assert lane.certified_non_reasoning is True, alias
assert lane.max_tokens == 1400, alias
# Empty/unset (provider default) and real efforts must NOT certify.
for not_disabled in ("", None, "low", "high", True):
lane = _resolve({**base, "reasoning_effort": not_disabled})
assert lane.certified_non_reasoning is False, not_disabled
assert lane.max_tokens is None, not_disabled