From 2b09ecea76346a371bc7886eeb66d88d2489a50e Mon Sep 17 00:00:00 2001 From: JD Davis Date: Sat, 11 Jul 2026 15:25:15 +0000 Subject: [PATCH] refactor(proxy): isolate image compression policy (#1958) ## Description Extracts image-compression gating and tag stamping into a pure policy module while preserving the public `ImageCompressionDecision.decide` API used by handlers. This keeps the frozen decision value type separate from the canonical precedence rules it wraps. Closes # ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [x] Code refactoring (no functional changes) ## Changes Made - Added `headroom.proxy.image_compression_policy` for pure image-compression precedence and tag stamping helpers. - Updated `ImageCompressionDecision.decide` and `ImageCompressionDecision.apply_to_tags` to delegate to the extracted policy. - Added direct tests for the extracted policy boundary. - Kept the LiteLLM callback compatibility shim required for repo-wide type checking on fresh branches. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text python -m pytest tests/test_image_compression_policy.py tests/test_image_compression_decision.py tests/test_handler_outcome_tag_invariant.py tests/test_litellm_callback.py tests/test_compress_api.py::TestLiteLLMCallback -q 34 passed in 6.77s python -m ruff check . All checks passed! python -m ruff format --check . 1095 files already formatted python -m mypy headroom --ignore-missing-imports Success: no issues found in 409 source files gitleaks protect --staged --no-banner --redact no leaks found ``` ## Real Behavior Proof - Environment: Windows, Python 3.13.13, local worktree based on `headroomlabs/main`. - Exact command / steps: ran focused image compression policy/decision tests, handler outcome tag invariant tests, LiteLLM callback tests, Ruff lint/format checks, mypy over `headroom`, and staged gitleaks scan. - Observed result: all local checks passed; staged secret scan found no leaks. - Not tested: full CI matrix and deployment flows; those are covered by GitHub Actions. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) N/A ## Additional Notes Documentation and changelog updates are not applicable for this internal refactor. GitHub reported existing Dependabot alerts on the default branch during push; this PR does not change dependencies, and the staged secret scan is clean. --- headroom/proxy/image_compression_decision.py | 39 +++++-------- headroom/proxy/image_compression_policy.py | 56 +++++++++++++++++++ tests/test_image_compression_policy.py | 58 ++++++++++++++++++++ 3 files changed, 129 insertions(+), 24 deletions(-) create mode 100644 headroom/proxy/image_compression_policy.py create mode 100644 tests/test_image_compression_policy.py diff --git a/headroom/proxy/image_compression_decision.py b/headroom/proxy/image_compression_decision.py index b558a0cd1..5a403524b 100644 --- a/headroom/proxy/image_compression_decision.py +++ b/headroom/proxy/image_compression_decision.py @@ -31,7 +31,10 @@ from collections.abc import Sequence from dataclasses import dataclass from typing import Any -from headroom.proxy.helpers import _headroom_bypass_enabled +from headroom.proxy.image_compression_policy import ( + apply_image_skip_reason, + decide_image_compression, +) @dataclass(frozen=True) @@ -84,29 +87,18 @@ class ImageCompressionDecision: messages Request messages. ``None`` and ``[]`` are equivalent. """ - bypass = _headroom_bypass_enabled(headers) - image_ok = bool(getattr(config, "image_optimize", False)) - has_msgs = bool(messages) - - if bypass: - reason: str | None = "bypass_header" - should = False - elif not image_ok: - reason = "image_optimize_disabled" - should = False - elif not has_msgs: - reason = "no_messages" - should = False - else: - reason = None - should = True + decision = decide_image_compression( + headers=headers, + image_optimize_enabled=bool(getattr(config, "image_optimize", False)), + has_messages=bool(messages), + ) return cls( - should_compress=should, - passthrough_reason=reason, - bypass_header_set=bypass, - image_optimize_enabled=image_ok, - has_messages=has_msgs, + should_compress=decision.should_compress, + passthrough_reason=decision.passthrough_reason, + bypass_header_set=decision.bypass_header_set, + image_optimize_enabled=decision.image_optimize_enabled, + has_messages=decision.has_messages, ) def apply_to_tags(self, tags: dict[str, str]) -> None: @@ -121,5 +113,4 @@ class ImageCompressionDecision: ``memory_skip_reason``, ``image_skip_reason``) for full dashboard slicing. """ - if self.passthrough_reason is not None: - tags["image_skip_reason"] = self.passthrough_reason + apply_image_skip_reason(tags, self.passthrough_reason) diff --git a/headroom/proxy/image_compression_policy.py b/headroom/proxy/image_compression_policy.py new file mode 100644 index 000000000..e336c6438 --- /dev/null +++ b/headroom/proxy/image_compression_policy.py @@ -0,0 +1,56 @@ +"""Pure image-compression decision policy helpers.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from headroom.proxy.helpers import _headroom_bypass_enabled + + +@dataclass(frozen=True) +class ImageCompressionPolicyResult: + """Raw image-compression gate result before wrapping in public value types.""" + + should_compress: bool + passthrough_reason: str | None + bypass_header_set: bool + image_optimize_enabled: bool + has_messages: bool + + +def decide_image_compression( + *, + headers: Any, + image_optimize_enabled: bool, + has_messages: bool, +) -> ImageCompressionPolicyResult: + """Compute the canonical image-compression gate.""" + bypass = _headroom_bypass_enabled(headers) + + if bypass: + reason: str | None = "bypass_header" + should = False + elif not image_optimize_enabled: + reason = "image_optimize_disabled" + should = False + elif not has_messages: + reason = "no_messages" + should = False + else: + reason = None + should = True + + return ImageCompressionPolicyResult( + should_compress=should, + passthrough_reason=reason, + bypass_header_set=bypass, + image_optimize_enabled=image_optimize_enabled, + has_messages=has_messages, + ) + + +def apply_image_skip_reason(tags: dict[str, str], passthrough_reason: str | None) -> None: + """Stamp image skip reason into tags when present.""" + if passthrough_reason is not None: + tags["image_skip_reason"] = passthrough_reason diff --git a/tests/test_image_compression_policy.py b/tests/test_image_compression_policy.py new file mode 100644 index 000000000..870a92012 --- /dev/null +++ b/tests/test_image_compression_policy.py @@ -0,0 +1,58 @@ +"""Tests for pure image-compression decision policy helpers.""" + +from __future__ import annotations + +from headroom.proxy.image_compression_policy import ( + apply_image_skip_reason, + decide_image_compression, +) + + +def test_decide_image_compression_allows_happy_path() -> None: + decision = decide_image_compression( + headers={}, + image_optimize_enabled=True, + has_messages=True, + ) + + assert decision.should_compress is True + assert decision.passthrough_reason is None + + +def test_decide_image_compression_uses_canonical_precedence() -> None: + decision = decide_image_compression( + headers={"x-headroom-bypass": "true"}, + image_optimize_enabled=False, + has_messages=False, + ) + + assert decision.should_compress is False + assert decision.passthrough_reason == "bypass_header" + assert decision.bypass_header_set is True + assert decision.image_optimize_enabled is False + assert decision.has_messages is False + + +def test_decide_image_compression_reports_config_and_message_reasons() -> None: + disabled = decide_image_compression( + headers={}, + image_optimize_enabled=False, + has_messages=True, + ) + empty = decide_image_compression( + headers={}, + image_optimize_enabled=True, + has_messages=False, + ) + + assert disabled.passthrough_reason == "image_optimize_disabled" + assert empty.passthrough_reason == "no_messages" + + +def test_apply_image_skip_reason_stamps_only_when_skipping() -> None: + tags: dict[str, str] = {} + apply_image_skip_reason(tags, None) + assert tags == {} + + apply_image_skip_reason(tags, "no_messages") + assert tags == {"image_skip_reason": "no_messages"}