From 75e32f20f7c5461a813a604689bd4aaa7cbe4739 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 22 May 2026 14:47:06 +0000 Subject: [PATCH] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- .../core/inference/context_compaction.py | 21 ++---- .../backend/tests/test_context_compaction.py | 73 ++++++++++++------- 2 files changed, 52 insertions(+), 42 deletions(-) diff --git a/studio/backend/core/inference/context_compaction.py b/studio/backend/core/inference/context_compaction.py index 3920ee3e3e..8428d0cd53 100644 --- a/studio/backend/core/inference/context_compaction.py +++ b/studio/backend/core/inference/context_compaction.py @@ -117,9 +117,7 @@ class CompactStrategy(ABC): """Interface for context-compaction strategies.""" @abstractmethod - def compact( - self, messages: list[dict], budget_tokens: int - ) -> list[dict]: + def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]: """Return a (possibly shorter) list of messages within ``budget_tokens``. Returns ``messages`` unchanged when no compaction is needed or possible. @@ -130,9 +128,7 @@ class CompactStrategy(ABC): class NoCompact(CompactStrategy): """Passthrough strategy. Returns ``messages`` unchanged.""" - def compact( - self, messages: list[dict], budget_tokens: int - ) -> list[dict]: + def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]: return list(messages) @@ -147,9 +143,7 @@ class SlidingWindowCompact(CompactStrategy): within ``budget_tokens`` or when there is nothing left to drop. """ - def __init__( - self, keep_recent: int = 2, compact_threshold: float = 0.85 - ) -> None: + def __init__(self, keep_recent: int = 2, compact_threshold: float = 0.85) -> None: if keep_recent < 0: raise ValueError("keep_recent must be >= 0") if not (0.0 < compact_threshold <= 1.0): @@ -157,9 +151,7 @@ class SlidingWindowCompact(CompactStrategy): self.keep_recent = keep_recent self.compact_threshold = compact_threshold - def compact( - self, messages: list[dict], budget_tokens: int - ) -> list[dict]: + def compact(self, messages: list[dict], budget_tokens: int) -> list[dict]: if budget_tokens <= 0 or not messages: return list(messages) @@ -224,10 +216,7 @@ class SlidingWindowCompact(CompactStrategy): # Drop oldest-first until under threshold or nothing left. dropped: set[int] = set() for i in droppable: - kept = [ - m for j, m in enumerate(messages) - if j not in dropped and j != i - ] + kept = [m for j, m in enumerate(messages) if j not in dropped and j != i] if estimate_tokens(kept) <= threshold: dropped.add(i) break diff --git a/studio/backend/tests/test_context_compaction.py b/studio/backend/tests/test_context_compaction.py index f95c0499fe..03e688e78a 100644 --- a/studio/backend/tests/test_context_compaction.py +++ b/studio/backend/tests/test_context_compaction.py @@ -65,11 +65,13 @@ class TestEstimateTokens: _msg( "assistant", content = "", - tool_calls = [{ - "id": "c1", - "type": "function", - "function": {"name": "web_search", "arguments": '{"q":"hi"}'}, - }], + tool_calls = [ + { + "id": "c1", + "type": "function", + "function": {"name": "web_search", "arguments": '{"q":"hi"}'}, + } + ], ) ] # 10 char arguments -> 2 tokens. @@ -88,7 +90,11 @@ class TestNoCompact: class TestSlidingWindowUnderBudget: def test_no_op_when_already_fits(self): - msgs = [_msg("system", "be concise"), _msg("user", "hi"), _msg("assistant", "bye")] + msgs = [ + _msg("system", "be concise"), + _msg("user", "hi"), + _msg("assistant", "bye"), + ] out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 1000) assert out == msgs @@ -164,16 +170,23 @@ class TestSlidingWindowToolPairs: _msg( "assistant", content = "", - tool_calls = [{ - "id": "call_42", - "type": "function", - "function": { - "name": "web_search", - "arguments": '{"q":"x"}', - }, - }], + tool_calls = [ + { + "id": "call_42", + "type": "function", + "function": { + "name": "web_search", + "arguments": '{"q":"x"}', + }, + } + ], + ), + _msg( + "tool", + content = "result for x", + tool_call_id = "call_42", + name = "web_search", ), - _msg("tool", content = "result for x", tool_call_id = "call_42", name = "web_search"), _long("assistant", 800), _long("user", 800), _long("assistant", 800), @@ -185,7 +198,8 @@ class TestSlidingWindowToolPairs: # Force aggressive compaction. out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 50) kept_assistant_with_calls = [ - m for m in out + m + for m in out if m.get("role") == "assistant" and isinstance(m.get("tool_calls"), list) ] kept_tool_msgs = [m for m in out if m.get("role") == "tool"] @@ -215,22 +229,27 @@ class TestSlidingWindowToolPairs: _msg( "assistant", content = "", - tool_calls = [{ - "id": "call_42", - "type": "function", - "function": { - "name": "web_search", - "arguments": '{"q":"x"}', - }, - }], + tool_calls = [ + { + "id": "call_42", + "type": "function", + "function": { + "name": "web_search", + "arguments": '{"q":"x"}', + }, + } + ], ), _msg("tool", content = "result", tool_call_id = "call_42", name = "web_search"), _msg("user", "thanks"), ] out = SlidingWindowCompact(keep_recent = 2).compact(msgs, budget_tokens = 50) ids = [ - (m.get("role"), m.get("tool_call_id") or - (m.get("tool_calls") and m["tool_calls"][0].get("id"))) + ( + m.get("role"), + m.get("tool_call_id") + or (m.get("tool_calls") and m["tool_calls"][0].get("id")), + ) for m in out ] assert ("assistant", "call_42") in ids @@ -250,11 +269,13 @@ class TestStrategyRegistry: class TestConstructorValidation: def test_negative_keep_recent_raises(self): import pytest + with pytest.raises(ValueError): SlidingWindowCompact(keep_recent = -1) def test_invalid_threshold_raises(self): import pytest + with pytest.raises(ValueError): SlidingWindowCompact(compact_threshold = 0.0) with pytest.raises(ValueError):