fix(providers): guard compaction against token estimate drift

src/conductor/providers/_pydantic_ai/compaction.py

@@ -94,6 +94,25 @@ async def _estimate_context_tokens(
     )
 
 
+def _count_utf8_bytes(text: str) -> int:
+    """Return a one-byte-per-token safety bound for token-dense text."""
+    return len(text.encode("utf-8"))
+
+
+def _estimate_conservative_context_tokens(
+    messages: list[ModelMessage],
+    model_request_parameters: ModelRequestParameters | None,
+) -> int:
+    """Estimate context with a one-byte-per-token safety bound."""
+    from pydantic_ai_harness.compaction import estimate_context_tokens
+
+    return estimate_context_tokens(
+        messages,
+        tokenizer=_count_utf8_bytes,
+        model_request_parameters=model_request_parameters,
+    )
+
+
 def _estimate_after_compaction_tokens(
     before_messages: list[ModelMessage],
     after_messages: list[ModelMessage],
@@ -254,28 +273,59 @@ async def before_model_request(
         if self._disabled:
             return request_context
 
-        # Zone (a): gate measurement. A broken estimate says nothing about
-        # the compaction path, so this warns and skips compaction for this
-        # request only — no errored event, no disable latch.
+        before_messages = list(request_context.messages)
+        conservative_estimate: int | None = None
         try:
-            before_messages = list(request_context.messages)
             before_estimate = await _estimate_context_tokens(
                 before_messages,
                 request_context.model_request_parameters,
             )
         except Exception:  # noqa: BLE001 - estimation must never fail the run
             logger.warning(
-                "Compaction gate measurement failed for agent %r; "
-                "skipping compaction for this request.",
+                "Compaction gate measurement failed for agent %r; using the conservative fallback.",
                 self._config.agent_name,
                 exc_info=True,
             )
+            try:
+                conservative_estimate = _estimate_conservative_context_tokens(
+                    before_messages,
+                    request_context.model_request_parameters,
+                )
+            except Exception:  # noqa: BLE001 - estimation must never fail the run
+                logger.warning(
+                    "Compaction fallback measurement failed for agent %r; "
+                    "skipping compaction for this request.",
+                    self._config.agent_name,
+                    exc_info=True,
+                )
+                return request_context
+            before_estimate = conservative_estimate
+        else:
+            try:
+                conservative_estimate = _estimate_conservative_context_tokens(
+                    before_messages,
+                    request_context.model_request_parameters,
+                )
+            except Exception:  # noqa: BLE001 - primary estimate remains usable
+                logger.warning(
+                    "Compaction safety measurement failed for agent %r; "
+                    "using the primary estimate.",
+                    self._config.agent_name,
+                    exc_info=True,
+                )
+
+        # The primary estimate drives the reserve-based trigger. The byte bound
+        # only guards the hard window, avoiding premature compaction of ordinary
+        # text while catching suffixes the four-characters heuristic undercounts.
+        exceeds_trigger = before_estimate > self._config.trigger_tokens
+        may_exceed_window = (
+            conservative_estimate is not None
+            and conservative_estimate >= self._config.window_tokens
+        )
+        if not exceeds_trigger and not may_exceed_window:
             return request_context
-
-        # Gate on the token estimate, not the message count: one large
-        # prompt can exceed the trigger with nothing to drop.
-        if before_estimate <= self._config.trigger_tokens:
-            return request_context
+        if conservative_estimate is not None and may_exceed_window:
+            before_estimate = max(before_estimate, conservative_estimate)
 
         self._on_before(estimate=before_estimate, messages_before=len(before_messages))
         for tier in self._tier_wrappers:

tests/test_providers/test_pydantic_ai_compaction.py

@@ -332,6 +332,60 @@ async def test_gate_measures_once_via_wrapper_estimate(self) -> None:
         assert len(result.messages) == 1
 
 
+class TestKnownWindowSafety:
+    """Requirement: known context windows are hard pre-request boundaries."""
+
+    @pytest.mark.asyncio
+    async def test_token_dense_suffix_compacts_before_known_window_overflow(self) -> None:
+        # Requirement: provider usage plus token-dense suffix growth must compact before
+        # the next request can exceed the known context window.
+        capability = build_tiered_compaction(
+            _make_config(trigger_tokens=121_000, target_tokens=100_000)
+        )
+        messages: list[Any] = [
+            ModelResponse(
+                parts=[TextPart(content="compaction complete")],
+                usage=RequestUsage(input_tokens=74_499, output_tokens=0),
+            )
+        ]
+        for index in range(30):
+            messages.append(
+                ModelRequest(parts=[UserPromptPart(content=f"turn-{index}-" + "!" * 4_193)])
+            )
+        request_context = _request_context_with_messages(messages)
+
+        result = await capability.before_model_request(_make_run_context(), request_context)
+
+        assert len(result.messages) < len(messages)
+        assert len(result.messages) == 21
+
+    @pytest.mark.asyncio
+    async def test_estimator_failure_uses_safe_fallback_before_large_request(self) -> None:
+        # Requirement: a failed primary estimate must not bypass compaction when an
+        # independent conservative estimate shows the request can exceed the known window.
+        inner = AsyncMock()
+        compacted = _request_context_with_messages(
+            [ModelRequest(parts=[UserPromptPart(content="compacted")])]
+        )
+        inner.before_model_request = AsyncMock(return_value=compacted)
+        capability = _FailOpenCompactionWrapper(
+            inner,
+            config=_make_config(trigger_tokens=121_000, target_tokens=100_000),
+        )
+        request_context = _request_context_with_messages(
+            [ModelRequest(parts=[UserPromptPart(content="!" * 200_288)])]
+        )
+
+        with patch(
+            "conductor.providers._pydantic_ai.compaction._estimate_context_tokens",
+            new=AsyncMock(side_effect=RuntimeError("estimator exploded")),
+        ):
+            result = await capability.before_model_request(_make_run_context(), request_context)
+
+        inner.before_model_request.assert_called_once()
+        assert result is compacted
+
+
 class TestTierFallback:
     """Requirement: a failing non-final tier still yields to the final tier."""
 
@@ -920,9 +974,9 @@ def callback(event_type: str, data: dict[str, Any]) -> None:
         assert any(e[0] == "agent_compaction_complete" for e in events)
 
     @pytest.mark.asyncio
-    async def test_gate_measurement_failure_skips_without_latch_or_event(self) -> None:
-        # A failing gate estimate warns, returns the
-        # original context, emits no event, and does not engage the latch.
+    async def test_gate_measurement_failure_uses_fallback_without_latch(self) -> None:
+        # Requirement: a failed primary estimate uses the conservative fallback
+        # without disabling compaction for later requests.
         events: list[tuple[str, dict[str, Any]]] = []
 
         def callback(event_type: str, data: dict[str, Any]) -> None:
@@ -942,7 +996,10 @@ def callback(event_type: str, data: dict[str, Any]) -> None:
 
         assert result is request_context
         assert capability._disabled is False  # type: ignore[attr-defined]
-        assert not events
+        assert [event_type for event_type, _ in events] == [
+            "agent_compaction_start",
+            "agent_compaction_complete",
+        ]
 
     @pytest.mark.asyncio
     async def test_telemetry_failure_keeps_compacted_result_without_latch(self) -> None: