diff --git a/AGENTS.md b/AGENTS.md index 61605d14..51c6124e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -228,7 +228,7 @@ step-by-step checklist. - **Hardened dashboard request surface** (issue #397): three independent layers, none individually load-bearing. (1) `web/auth.py::OriginHostGuard`, a pure-ASGI middleware (`app.add_middleware(...)`, not `BaseHTTPMiddleware`/`@app.middleware("http")` — neither sees WebSocket scopes) validating `Host` (required, must name the bound machine) and `Origin` (only checked when present — httpx/curl/`conductor gate respond` send none, and that path must keep working with no extra setup) on every HTTP and WebSocket request. `CONDUCTOR_WEB_ALLOW_ORIGINS` (comma-separated full origins) is the dev-server escape hatch, additive only. (2) A per-run token, minted automatically (`mint_token()`) so the protected configuration is the default; `CONDUCTOR_GATE_TOKEN` overrides it (`resolve_expected_token`). Required on every mutating route (`/api/stop`, `/api/kill`, `/api/resume`, `/api/gate-respond`, `/api/guidance`) and on the `/ws` handshake — the single auth point for the socket, since an unauthenticated connection is closed (`websocket.close`, code 1008) in reply to `websocket.connect`, before `accept()`, so it can never send *any* message type. This is strictly stronger than the pre-#397 per-message check, which browsers could never satisfy at all. Read-only routes (`/api/state`, `/api/info`, `/api/logs`, `/api/gate-status`, `/api/files/*`, the whole replay app) stay unauthenticated, protected by origin/host only. (3) `Content-Type: application/json` required on every mutating route (415 otherwise), including the bodyless control POSTs. **Token discovery:** `WebDashboard.start()` writes a `0600` file (POSIX; on Windows the mode is not honoured and the file relies on the user-profile NTFS ACL instead) at `~/.conductor/runs/dashboard-.token` once the port resolves (works for both `--web` and `--web-bg`, since `cli/run.py` calls `start()`/`stop()` on both the run and resume paths); `stop()` removes it. `conductor gate respond`, `conductor guide`, and `conductor stop`'s graceful-kill rung all resolve a token via the shared `resolve_cli_token(port, token)`: `--token` > `CONDUCTOR_GATE_TOKEN` > the token file. See `docs/cli-reference.md` (Environment Variables, and the Authentication sections under `conductor gate respond` / `conductor guide`) and the `web/` bullets above for the full mechanism. - **MCP server exposure** (`mcp:` workflow block, `conductor mcp serve`, issue #432): `workflow.mcp` (`config/schema.py::McpConfig`) is a typed, `extra="forbid"` block — `expose` (default `true`; every workflow is a candidate for MCP tool exposure with no editing required, DD4), `mode` (`async`/`sync`/`auto`; the default a generated tool's omitted `_wait_seconds` resolves to), `read_only` / `destructive` (surfaced as the generated tool's `readOnlyHint`/`destructiveHint` annotations), and `estimated_minutes` (a client-side hint, must be positive). `conductor validate` reports an unknown key inside it as a schema error, not silence (FR11) — it cannot ride on the existing untyped `metadata: dict`. See `examples/mcp-serve.yaml` and `docs/mcp-server.md` (the user-facing guide: host configuration, the exposure ladder, toolsets, the run lifecycle, and a dedicated *Limits* section for DD5/DD9/DD11/DD12/R4) and `src/conductor/mcp/serve/` above for the server that reads it. **R1 — this feature's terminal run record is a scope change to `conductor status` and `conductor fleet list`, not an MCP-only addition**: every run — MCP-launched or not — now writes a completion tombstone, so both commands (and the Fleet TUI's History screen) gained a completed-runs section as a side effect, with `--live` restoring the exact pre-change scope. See the `fleet/records.py` (`TerminalRunRecord`) bullet above and the `CHANGELOG.md` entry for the full description of what changed. -- **Context compaction**: Always-on client-side context window compaction for the `claude` and `openai` providers. Compaction is triggered proactively using the reserve-based formula `trigger = window - (output_limit + buffer)` and targets a clamped 55% hysteresis ceiling. In this formula, the `output_limit` resolves to the minimum of the effective `max_tokens` sent to the API (from settings or defaults) and the model output cap (from provider-cap). The tool-output-derived `buffer` is calculated as `2 * ceil(max_chars / 4) + 15,000` tokens as a heuristic for worst-case tool results. Resolution prioritizes cascades: provider-advertised metadata (with full pagination for Anthropic and a vendor-field parser for OpenAI-compatible endpoints), public registry, and conservative fallback. The wrapper operates in a fail-open manner, disabling itself for the rest of the agent execution upon outer failure. Compaction events are emitted to the dashboard and console. A summarizing compaction step runs a nested model call that consumes a request slot from the agent's `max_agent_iterations` budget, which doesn't get refunded. The dashboard context bar relies on provider-only limits and may disagree with the compaction window. +- **Context compaction**: Always-on client-side context window compaction for the `claude` and `openai` providers. Compaction is triggered proactively using the reserve-based formula `trigger = window - (output_limit + buffer)` and targets a clamped 55% hysteresis ceiling. In this formula, the `output_limit` resolves to the minimum of the effective `max_tokens` sent to the API (from settings or defaults) and the model output cap (from provider-cap). The tool-output-derived `buffer` is calculated as `2 * ceil(max_chars / 4) + 15,000` tokens as a heuristic for worst-case tool results. Resolution prioritizes cascades: provider-advertised metadata (with full pagination for Anthropic and a vendor-field parser for OpenAI-compatible endpoints), public registry, and conservative fallback. A second, density-calibrated estimate guards the hard window against token-dense content (CJK, base64, minified data) the ~4-chars-per-token heuristic undercounts: it matches the heuristic on ordinary prose, so it never compacts a history that is merely large, and when it fires the tier chain is driven directly against that measurement rather than delegated to the inner strategy's own (heuristic) gate. Token telemetry stays on the primary scale — the density value travels as `density_tokens` and `trigger_reason` on the start event, with `degraded_estimators` / `still_over_window` on the complete event and an `agent_compaction_skipped` event when both estimators fail. The wrapper operates in a fail-open manner, disabling itself for the rest of the agent execution upon outer failure. Compaction events are emitted to the dashboard and console. A summarizing compaction step runs a nested model call that consumes a request slot from the agent's `max_agent_iterations` budget, which doesn't get refunded. The dashboard context bar relies on provider-only limits and may disagree with the compaction window. ### Debugging `--web-bg` failures diff --git a/CHANGELOG.md b/CHANGELOG.md index f3a369a1..6d757ffc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 entirely. See [`examples/claude-agent-sdk-setting-sources.yaml`](examples/claude-agent-sdk-setting-sources.yaml). +### Fixed + +- **Context compaction window guard against token-dense drift** (#507) — the + `claude` / `openai` providers' compaction trigger anchors on + provider-reported token usage and estimates everything after the anchor + with a ~4-characters-per-token heuristic, which undercounts token-dense + content (CJK and other non-Latin scripts, base64, hex, minified data) by + 2-4x. A dense suffix could therefore grow the real request past a known + context window while the trigger estimate stayed below the threshold, and + the provider rejected the request with `context_length_exceeded`. A second, + density-calibrated estimate now guards the hard window: it matches the + primary heuristic on ordinary prose, counts text with a substantial + non-ASCII share at ~1 token per character, and whitespace-poor ASCII blobs + at ~2 characters per token, so it fires only on genuinely dense content — + never on a history that is merely large. When it fires, the tier chain is + driven directly against that measurement (the inner strategy's own gate + would re-measure with the same heuristic that under-counted the content + and no-op), until the estimate is back under the target. Telemetry stays on + the token scale: `agent_compaction_start` gains `trigger_reason` + (`"trigger"` / `"window_guard"`) and a separate `density_tokens` field + instead of overloading `tokens_before`, and `agent_compaction_complete` + gains `degraded_estimators` and `still_over_window` so a guard compaction + that could not get back under the window reads as degraded, not as false + success. A failed primary measurement falls back to an independent + density-calibrated estimate that shares no code with it, and a double + failure is reported as a new `agent_compaction_skipped` event + (`reason: "estimate_unavailable"`) rather than vanishing into stderr. See + [Workflow Syntax → Context Compaction](docs/workflow-syntax.md#context-compaction). + ## [0.1.36](https://github.com/microsoft/conductor/compare/v0.1.35...v0.1.36) - 2026-09-02 ### Added diff --git a/docs/workflow-syntax.md b/docs/workflow-syntax.md index e6603705..8de7b9ce 100644 --- a/docs/workflow-syntax.md +++ b/docs/workflow-syntax.md @@ -2425,6 +2425,14 @@ Below is how these values resolve in practice for different configurations using When the reserve (output limit plus effective tool buffer) leaves no viable headroom below the window, compaction is disabled for the agent execution rather than armed with a degenerate threshold. The `agent_compaction_config` event then carries `enabled: false` and a `disabled_reason`, so the condition is visible per run instead of surfacing as a one-shot log warning. To resolve this, lower `runtime.max_tokens` or `tool_output.max_chars`. +#### Window Guard Against Token-Dense Content + +The trigger is measured by the primary estimator: the provider's reported token usage for the history up to the most recent response, plus a ~4-characters-per-token heuristic for everything after it. That heuristic undercounts token-dense content — CJK and other non-Latin scripts, base64, hex, or minified data — by 2-4x, so a dense suffix can grow the real request past the known context window while the trigger estimate stays below the threshold. + +A second, density-calibrated estimate guards the hard window. It matches the primary heuristic on ordinary prose, counts text with a substantial non-ASCII share at ~1 token per character, and whitespace-poor ASCII blobs at ~2 characters per token. When that estimate reaches the known window, compaction runs even if the trigger never fired, and the tier chain is driven against the density-calibrated measurement until the history fits the target. Because the two estimates agree on ordinary text, the guard never compacts a history that is merely large. + +The start event reports which gate fired via `trigger_reason` (`"trigger"` or `"window_guard"`) and carries the density-calibrated value separately as `density_tokens`; `tokens_before` always stays the primary token estimate. + ### Compaction Tiers Conductor uses three sequential tiers to compress the history down to the target: @@ -2462,12 +2470,13 @@ All tokens consumed by summarizing compaction are added to the workflow's total ### Observability and Events -Compaction operates in a fail-open manner. If an error occurs during compaction, Conductor logs a warning, disables compaction for the rest of that agent's execution, and continues with the uncompacted history. +Compaction operates in a fail-open manner. If an error occurs during compaction, Conductor logs a warning, disables compaction for the rest of that agent's execution, and continues with the uncompacted history. A failed context measurement never disables anything: the primary estimate falls back to an independent density-calibrated one, and only when both fail is compaction skipped for that request alone, reported as `agent_compaction_skipped` with `reason: "estimate_unavailable"`. -Conductor emits three event types to track compaction: +Conductor emits four event types to track compaction: * `agent_compaction_config`: Emitted once at the start of agent execution to log resolved window and limit values. -* `agent_compaction_start`: Emitted when context size exceeds the trigger threshold and compaction begins. -* `agent_compaction_complete`: Emitted when compaction completes, detailing token savings or errors. +* `agent_compaction_start`: Emitted when compaction begins. `trigger_reason` names the gate that fired (`"trigger"` or `"window_guard"`), and `density_tokens` carries the density-calibrated estimate alongside the primary-scale `tokens_before`. +* `agent_compaction_complete`: Emitted when compaction completes, detailing token savings or errors. Degraded outcomes are named rather than hidden: `degraded_tiers` for recovered tier failures, `degraded_estimators` for lost measurements, `still_over_trigger` when the history remains above the trigger, and `still_over_window` when a window-guard compaction could not get back below the known window. +* `agent_compaction_skipped`: Emitted when compaction did not run because the context size could not be measured at all. ### Dashboard Caveat diff --git a/src/conductor/cli/run.py b/src/conductor/cli/run.py index c314e51a..2fcbaf18 100644 --- a/src/conductor/cli/run.py +++ b/src/conductor/cli/run.py @@ -1242,13 +1242,19 @@ def on_event(self, event: WorkflowEvent) -> None: messages_after = d.get("messages_after", 0) elapsed = d.get("elapsed", 0.0) degraded_tiers = d.get("degraded_tiers") or [] + degraded_estimators = d.get("degraded_estimators") or [] still_over_trigger = d.get("still_over_trigger", False) - if degraded_tiers or still_over_trigger: + still_over_window = d.get("still_over_window", False) + if degraded_tiers or degraded_estimators or still_over_trigger or still_over_window: reasons: list[str] = [] if degraded_tiers: reasons.append(f"tier(s) degraded: {', '.join(degraded_tiers)}") + if degraded_estimators: + reasons.append(f"estimator(s) degraded: {', '.join(degraded_estimators)}") if still_over_trigger: reasons.append("history remains above the trigger") + if still_over_window: + reasons.append("history remains above the known context window") verbose_log( styled( " WARNING: context compacted for '[bold]{}[/bold]': " @@ -1277,6 +1283,17 @@ def on_event(self, event: WorkflowEvent) -> None: ) ) + elif t == "agent_compaction_skipped": + verbose_log( + styled( + " WARNING: compaction skipped for '[bold]{}[/bold]' ({}) — " + "context size could not be measured for this request", + d.get("agent_name", "?"), + d.get("reason", "unknown"), + ), + style="yellow", + ) + elif t == "guidance_received": pending = d.get("pending", 1) verbose_log( diff --git a/src/conductor/providers/_pydantic_ai/compaction.py b/src/conductor/providers/_pydantic_ai/compaction.py index 05ddff6a..2abce151 100644 --- a/src/conductor/providers/_pydantic_ai/compaction.py +++ b/src/conductor/providers/_pydantic_ai/compaction.py @@ -3,13 +3,18 @@ This module assembles the harness's :class:`TieredCompaction` into a conductor wrapper that: -1. Gates compaction on a reserve-based token trigger (window minus the output - limit minus the tool buffer). +1. Gates compaction on two independent measurements: a reserve-based token + trigger (window minus the output limit minus the tool buffer), measured by + the primary estimator, and a hard context-window guard, measured by a + density-calibrated safety estimate for content the primary heuristic + undercounts. 2. Wraps each escalation tier individually so a failing LLM summarizer still yields to the deterministic sliding-window fallback. -3. Fails open: any unexpected error in the gate or tier chain is logged and the - original request context is returned unchanged, so a compaction bug never - aborts a workflow run. +3. Fails open: a failed primary measurement falls back to an independent + density-calibrated estimate; only when both estimators fail is the original + request context returned unchanged (with an ``agent_compaction_skipped`` + event). Any unexpected error in the tier chain is likewise logged and + returned unchanged, so a compaction bug never aborts a workflow run. 4. Emits ``agent_compaction_start`` / ``agent_compaction_complete`` events through the per-execute callback so the console, JSONL log, and dashboard can observe compaction activity. @@ -19,17 +24,19 @@ import logging import time -from dataclasses import dataclass +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, replace from typing import Any from pydantic_ai.capabilities import AbstractCapability -from pydantic_ai.messages import ModelMessage +from pydantic_ai.messages import ModelMessage, ModelResponse from pydantic_ai.models import ModelRequestContext, ModelRequestParameters from pydantic_ai.tools import RunContext from conductor.providers._pydantic_ai.events import ( emit_compaction_complete, emit_compaction_complete_error, + emit_compaction_skipped, emit_compaction_start, ) @@ -81,7 +88,9 @@ async def _estimate_context_tokens( """Primary token estimator using the harness helper. Counts message parts, instructions, and conservative tool-schema overhead - so the gate measures the same quantity the inner tiers measure. + so the gate measures the same quantity the inner tiers measure. ``async`` + by convention for provider operations and to keep the estimator call sites + uniform; the underlying harness call is synchronous. """ from pydantic_ai_harness.compaction import ( estimate_context_tokens, @@ -94,6 +103,103 @@ async def _estimate_context_tokens( ) +_DENSITY_SAMPLE_CHARS = 4_096 +"""Bounded leading sample used for the whitespace-density check.""" + +_WHITESPACE_CHARS = " \t\n\r\f\v" + + +def _density_text_token_bound(text: str) -> int: + """Estimate a token count that tracks token density instead of assuming prose. + + The ~4-characters-per-token heuristic is accurate for ordinary prose but + undercounts token-dense content by 2-4x, which is what lets a dense + history drift past a known context window while staying below the + compaction trigger. This bound keeps the heuristic for ordinary text and + escalates only for text that is measurably dense: + + - text with a substantial non-ASCII share (CJK, non-Latin scripts, emoji) + tokenizes near one token per character, so the bound is the character + count; + - ASCII text with almost no whitespace (base64, hex, minified data) + tokenizes near two characters per token, so the bound is half the + character count. + + The result is in TOKENS and matches the primary heuristic on ordinary + prose, so it can be compared against the context window without firing on + histories that are merely large. Multi-token-per-character sequences + (rare emoji runs) remain a known residual undercount; the bounded leading + sample keeps the classification cost flat for very large parts. + """ + chars = len(text) + if chars == 0: + return 0 + sample = text[:_DENSITY_SAMPLE_CHARS] + non_ascii = len(sample) - len(sample.encode("ascii", "ignore")) + if non_ascii * 10 >= len(sample): + return chars + whitespace = sum(sample.count(char) for char in _WHITESPACE_CHARS) + if whitespace * 10 < len(sample): + return chars // 2 + 1 + return chars // 4 + + +async def _estimate_context_tokens_density( + messages: list[ModelMessage], + model_request_parameters: ModelRequestParameters | None, +) -> int: + """Density-calibrated safety estimate of the context size, in tokens. + + Uses the same harness anchoring as the primary estimator but measures the + post-anchor suffix with :func:`_density_text_token_bound`, so the result + tracks the primary estimate on ordinary prose and rises up to ~4x above + it on token-dense suffixes. It is comparable against the context window + and the compaction target; it is never mixed into token telemetry as a + substitute for the primary estimate. + """ + from pydantic_ai_harness.compaction import estimate_context_tokens + + return estimate_context_tokens( + messages, + tokenizer=_density_text_token_bound, + model_request_parameters=model_request_parameters, + ) + + +async def _estimate_context_tokens_independent(messages: list[ModelMessage]) -> int: + """Density-calibrated context estimate sharing no code with the primary path. + + The primary and density-calibrated estimators both funnel through the + harness's text collection, so a bug there (a part whose ``str(...)`` + raises, a malformed usage anchor) fails them together. This fallback + walks the message list directly with attribute-level access and a + per-part guard, so one malformed part degrades the estimate instead of + raising. It is rougher than the harness path (no instruction or + tool-schema accounting) and is used only when the primary measurement + itself failed. + """ + anchor_tokens = 0 + anchor_index = -1 + for index in range(len(messages) - 1, -1, -1): + message = messages[index] + usage = getattr(message, "usage", None) + input_tokens = getattr(usage, "input_tokens", 0) or 0 + if isinstance(message, ModelResponse) and input_tokens: + anchor_tokens = int(input_tokens) + int(getattr(usage, "output_tokens", 0) or 0) + anchor_index = index + break + suffix = 0 + for message in messages[anchor_index + 1 :]: + for part in getattr(message, "parts", []): + try: + content = getattr(part, "content", "") + text = content if isinstance(content, str) else str(content) + suffix += _density_text_token_bound(text) + except Exception: # noqa: BLE001 - one bad part must not zero the estimate + continue + return anchor_tokens + suffix + + def _estimate_after_compaction_tokens( before_messages: list[ModelMessage], after_messages: list[ModelMessage], @@ -103,13 +209,20 @@ def _estimate_after_compaction_tokens( ``estimate_context_tokens`` anchors on the most recent ``ModelResponse`` with provider-reported ``usage.input_tokens``. That anchoring response - survives compaction (every tier keeps the recent tail), so a naive - after-estimate still describes the pre-rewrite request and always reports - ``after == before`` — i.e. ``tokens_saved == 0`` — no matter how much - history was dropped. ``TieredCompaction._escalate`` compensates for this - internally by subtracting the tier's measured heuristic reclaim from its - anchored baseline; this helper mirrors that compensation so the telemetry - reports the same numbers the escalation loop acted on. + usually survives compaction (every tier keeps the recent tail), so a + naive after-estimate still describes the pre-rewrite request and always + reports ``after == before`` — i.e. ``tokens_saved == 0`` — no matter how + much history was dropped. ``TieredCompaction._escalate`` compensates for + this internally by subtracting the tier's measured heuristic reclaim from + its anchored baseline; this helper mirrors that compensation so the + telemetry reports the same scale of numbers the escalation loop acted on. + Two caveats: the inputs must all be on the token scale (a baseline + produced by a different estimator invalidates the subtraction), and on a + window-guard compaction — where the anchor may be dropped and the real + reclaim is density-scale, which the heuristic under-measures — the + reported ``after`` overstates the remaining tokens and understates the + savings. That is the cheap direction: ``still_over_trigger`` stays honest + because it compares token-scale numbers against the token trigger. """ from pydantic_ai_harness.compaction import estimate_token_count @@ -156,15 +269,26 @@ async def compact( class _FailOpenCompactionWrapper(AbstractCapability[Any]): """Outer gate + fail-open wrapper around the tiered strategy. - The token gate lives here rather than in a separate capability: measuring - the context once per request (not once in a gate and again inside the - inner strategy's own trigger check) halves the estimator work. + Every request is measured twice, by design: the primary estimator + (provider usage anchor plus the ~4-characters-per-token heuristic) drives + the reserve-based trigger, and a density-calibrated safety estimate + guards the hard context window against content the heuristic undercounts. + On ordinary prose the two agree, so the safety estimate never fires on a + history that is merely large. Failure handling is zoned: - - **Gate measurement failure** — the before-estimate itself raised. Log a - warning and return the context unchanged; no event and no disable latch, - because a broken estimate says nothing about the compaction path. + - **Primary measurement failure** — fall back to an independent + density-calibrated estimate that shares no code with the primary path. + Compaction may still run and lifecycle events are emitted as usual; the + fallback value becomes ``tokens_before`` for this request. + - **Safety measurement failure** — the primary estimate remains usable, so + compaction proceeds on it alone and the complete event names + ``"density"`` in ``degraded_estimators``; the window guard is disarmed + for this request only. + - **Both estimators failed** — return the context unchanged with an + ``agent_compaction_skipped`` event; no disable latch, because a broken + estimate says nothing about the compaction path. - **Inner strategy failure** — log, emit an errored ``agent_compaction_complete``, engage the per-execution disable latch, and return the original context unchanged. @@ -189,7 +313,14 @@ def __init__( self._tier_wrappers: list[_TierWrapper] = tier_wrappers or [] self._disabled = False - def _on_before(self, estimate: int, messages_before: int) -> None: + def _on_before( + self, + estimate: int, + messages_before: int, + *, + trigger_reason: str, + density_estimate: int | None, + ) -> None: """Emit ``agent_compaction_start`` through the per-execute callback.""" emit_compaction_start( self._config.event_callback, @@ -204,6 +335,8 @@ def _on_before(self, estimate: int, messages_before: int) -> None: target_tokens=self._config.target_tokens, messages_before=messages_before, tokens_before=estimate, + trigger_reason=trigger_reason, + density_tokens=density_estimate, ) def _on_after( @@ -215,6 +348,8 @@ def _on_after( after_estimate: int, elapsed_seconds: float, degraded_tiers: list[str], + degraded_estimators: list[str], + still_over_window: bool, ) -> None: """Emit a success-shaped ``agent_compaction_complete`` event.""" emit_compaction_complete( @@ -231,6 +366,8 @@ def _on_after( elapsed=elapsed_seconds, degraded_tiers=degraded_tiers, still_over_trigger=after_estimate > self._config.trigger_tokens, + degraded_estimators=degraded_estimators, + still_over_window=still_over_window, ) def _on_error(self, exc: Exception) -> None: @@ -246,6 +383,29 @@ def _on_error(self, exc: Exception) -> None: context_window_source=self._config.window_source, ) + async def _drive_tiers_under_window_guard( + self, + messages: list[ModelMessage], + ctx: RunContext[Any], + density_of: Callable[[list[ModelMessage]], Awaitable[int]], + ) -> list[ModelMessage]: + """Drive the tier chain until the density-calibrated estimate fits the target. + + The inner strategy's own gate measures with the primary estimator — + the same heuristic that under-counted this content — so delegating to + it would no-op exactly where the window guard is needed. Driving the + tiers directly keeps the stop decision on the density-calibrated + scale. Escalation order and per-tier failure handling are unchanged: + a tier that raises is caught by its :class:`_TierWrapper` and named in + ``degraded_tiers``. + """ + compacted = list(messages) + for tier in self._tier_wrappers: + if await density_of(compacted) <= self._config.target_tokens: + break + compacted = await tier.compact(compacted, ctx) + return compacted + async def before_model_request( self, ctx: RunContext[Any], @@ -254,38 +414,127 @@ 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) + primary_estimate: int | None = None + density_estimate: int | None = None + degraded_estimators: list[str] = [] + try: - before_messages = list(request_context.messages) - before_estimate = await _estimate_context_tokens( + primary_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.", + "using the density-calibrated fallback.", self._config.agent_name, exc_info=True, ) + degraded_estimators.append("primary") + try: + density_estimate = await _estimate_context_tokens_independent(before_messages) + 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, + ) + emit_compaction_skipped( + self._config.event_callback, + agent_name=self._config.agent_name, + strategy="tiered", + model=self._config.model_name, + reason="estimate_unavailable", + ) + return request_context + else: + try: + density_estimate = await _estimate_context_tokens_density( + 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 only.", + self._config.agent_name, + exc_info=True, + ) + degraded_estimators.append("density") + + if primary_estimate is not None: + before_estimate = primary_estimate + elif density_estimate is not None: + before_estimate = density_estimate + else: + return request_context # pragma: no cover - the double-failure arm returned above + + # Two independent gates, two estimators. The reserve trigger uses the + # primary estimate (usage anchor + heuristic), which is accurate for + # ordinary text. The hard window guard uses the density-calibrated + # estimate, which matches the primary on ordinary prose and rises up + # to ~4x above it on token-dense content. The guard therefore fires + # only when the request may really overflow the known window — never + # merely because a history is large. ``before_estimate`` stays on the + # primary/fallback token scale: the density value is a gate input, + # reported separately as ``density_tokens``, never as token telemetry. + window_guard_tripped = ( + density_estimate is not None and density_estimate >= self._config.window_tokens + ) + trigger_tripped = before_estimate > self._config.trigger_tokens + if not trigger_tripped and not window_guard_tripped: 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 + # The density re-measurement used by the window-guard path: the shared + # harness-backed estimator normally, the independent fallback when the + # primary path is broken. + if primary_estimate is not None: + + async def density_of(messages: list[ModelMessage]) -> int: + return await _estimate_context_tokens_density( + messages, + request_context.model_request_parameters, + ) + + else: + + async def density_of(messages: list[ModelMessage]) -> int: + return await _estimate_context_tokens_independent(messages) - self._on_before(estimate=before_estimate, messages_before=len(before_messages)) + self._on_before( + estimate=before_estimate, + messages_before=len(before_messages), + trigger_reason="window_guard" if window_guard_tripped else "trigger", + density_estimate=density_estimate, + ) for tier in self._tier_wrappers: tier.failed = False start = time.monotonic() - # Zone (b): inner strategy. Fail open with the original context, emit - # the errored event, and latch the per-execution disable flag. + # Inner strategy. Fail open with the original context, emit the + # errored event, and latch the per-execution disable flag. When the + # window guard fired and the tier chain is available, drive it + # directly: delegating would re-gate on the same primary heuristic + # that under-counted this content and no-op. try: - result = await self._inner.before_model_request(ctx, request_context) + if window_guard_tripped and self._tier_wrappers: + # The tiers get the request's context, not the run's: a tier + # that resolves a model (the summarizing one) has to reach the + # same model the request is going to, mirroring the harness's + # own ``context_for_request``. + request_ctx = ( + ctx + if request_context.model is ctx.model + else replace(ctx, model=request_context.model) + ) + request_context.messages = await self._drive_tiers_under_window_guard( + before_messages, request_ctx, density_of + ) + result = request_context + else: + result = await self._inner.before_model_request(ctx, request_context) except Exception as exc: # noqa: BLE001 - compaction must never fail the run logger.warning( "Compaction failed for agent %r: %s. Continuing without compaction.", @@ -297,23 +546,36 @@ async def before_model_request( return request_context degraded_tiers = [t._tier_name for t in self._tier_wrappers if t.failed] + guard_path = window_guard_tripped and bool(self._tier_wrappers) + if guard_path and list(result.messages) == before_messages: + logger.error( + "Window guard fired for agent %r but compaction changed nothing; " + "the request may still exceed the known context window.", + self._config.agent_name, + ) - # Zone (c): after-telemetry. Compaction already happened, so a failure - # here must still return the compacted result — warn, emit nothing, - # and leave the latch off. + # After-telemetry. Compaction already happened, so a failure here must + # still return the compacted result — warn, emit nothing, and leave + # the latch off. try: + after_messages = list(result.messages) + still_over_window = guard_path and ( + await density_of(after_messages) >= self._config.window_tokens + ) after_estimate = _estimate_after_compaction_tokens( before_messages, - list(result.messages), + after_messages, before_estimate, ) self._on_after( before_messages=before_messages, - after_messages=list(result.messages), + after_messages=after_messages, before_estimate=before_estimate, after_estimate=after_estimate, elapsed_seconds=time.monotonic() - start, degraded_tiers=degraded_tiers, + degraded_estimators=degraded_estimators, + still_over_window=still_over_window, ) except Exception: # noqa: BLE001 - telemetry must never fail the run logger.warning( @@ -332,9 +594,10 @@ def build_tiered_compaction(config: CompactionConfig) -> AbstractCapability[Any] The stack is, from outside in: - 1. ``_FailOpenCompactionWrapper`` — owns the token gate (measured once per - request), catches unexpected errors, and returns the original context - unchanged when the inner strategy fails. + 1. ``_FailOpenCompactionWrapper`` — owns the two gates (primary trigger + and density-calibrated window guard, each measured per request), + catches unexpected errors, and returns the original context unchanged + when the inner strategy fails. 2. ``TieredCompaction`` — escalates through the three tiers. 3. Per-tier wrappers around ``ClearToolResults``, ``SummarizingCompaction``, and ``SlidingWindowCompaction`` so a non-final tier failure still proceeds diff --git a/src/conductor/providers/_pydantic_ai/events.py b/src/conductor/providers/_pydantic_ai/events.py index e8278d4c..cc08552c 100644 --- a/src/conductor/providers/_pydantic_ai/events.py +++ b/src/conductor/providers/_pydantic_ai/events.py @@ -359,11 +359,20 @@ def emit_compaction_start( target_tokens: int, messages_before: int, tokens_before: int, + trigger_reason: str = "trigger", + density_tokens: int | None = None, ) -> None: """Emit an ``agent_compaction_start`` event, swallowing callback errors. - This event fires immediately before compaction begins on a request whose - estimated context size is above the trigger threshold. + This event fires immediately before compaction begins. ``trigger_reason`` + names the gate that fired: ``"trigger"`` when the primary token estimate + exceeded the reserve-based trigger threshold, or ``"window_guard"`` when + the density-calibrated safety estimate reached the known context window + (which can happen well below the trigger on token-dense content). + ``tokens_before`` is always the primary-scale token estimate; the + density-calibrated value that fired the window guard is reported + separately as ``density_tokens`` (``None`` when that measurement was + unavailable). """ if event_callback is None: return @@ -383,6 +392,8 @@ def emit_compaction_start( "target_tokens": target_tokens, "messages_before": messages_before, "tokens_before": tokens_before, + "trigger_reason": trigger_reason, + "density_tokens": density_tokens, }, ) except Exception: @@ -404,6 +415,8 @@ def emit_compaction_complete( elapsed: float, degraded_tiers: list[str], still_over_trigger: bool, + degraded_estimators: list[str] | None = None, + still_over_window: bool = False, ) -> None: """Emit a success-shaped ``agent_compaction_complete`` event. @@ -413,6 +426,13 @@ def emit_compaction_complete( ``still_over_trigger`` reports that the post-compaction estimate remains above the trigger, so consumers can distinguish a full success from a degraded outcome instead of reading both as false success. + ``degraded_estimators`` names the measurements that were lost before + compaction ran — ``"primary"`` when the token gate fell back to the + density-calibrated estimate, ``"density"`` when the window guard was + disarmed for the request. ``still_over_window`` reports that a + window-guard compaction could not bring the density-calibrated estimate + back below the known context window — the request may still be rejected + by the provider. """ if event_callback is None: return @@ -435,6 +455,8 @@ def emit_compaction_complete( "errored": False, "degraded_tiers": degraded_tiers, "still_over_trigger": still_over_trigger, + "degraded_estimators": degraded_estimators or [], + "still_over_window": still_over_window, }, ) except Exception: @@ -483,3 +505,37 @@ def emit_compaction_complete_error( event_callback("agent_compaction_complete", payload) except Exception: logger.debug("Error in event_callback for agent_compaction_complete", exc_info=True) + + +def emit_compaction_skipped( + event_callback: EventCallback | None, + *, + agent_name: str, + strategy: str, + model: str, + reason: str, +) -> None: + """Emit an ``agent_compaction_skipped`` event, swallowing callback errors. + + This event fires when compaction did not run for a reason other than the + gate staying below its thresholds — currently only + ``reason="estimate_unavailable"``, when both the primary and the fallback + context measurements failed. It exists so a silently skipped safety gate + is visible in the JSONL log and dashboard instead of being + indistinguishable from a request that simply needed no compaction. + """ + if event_callback is None: + return + + try: + event_callback( + "agent_compaction_skipped", + { + "agent_name": agent_name, + "strategy": strategy, + "model": model, + "reason": reason, + }, + ) + except Exception: + logger.debug("Error in event_callback for agent_compaction_skipped", exc_info=True) diff --git a/tests/test_providers/test_pydantic_ai_compaction.py b/tests/test_providers/test_pydantic_ai_compaction.py index 96a69d92..0547426e 100644 --- a/tests/test_providers/test_pydantic_ai_compaction.py +++ b/tests/test_providers/test_pydantic_ai_compaction.py @@ -28,10 +28,14 @@ from conductor.providers._pydantic_ai.agent_builder import build_agent from conductor.providers._pydantic_ai.compaction import ( CompactionConfig, + _density_text_token_bound, + _estimate_context_tokens, + _estimate_context_tokens_density, _FailOpenCompactionWrapper, _TierWrapper, build_tiered_compaction, ) +from conductor.providers._pydantic_ai.compaction_window import resolve_compaction_plan @pytest.fixture(autouse=True) @@ -332,6 +336,294 @@ 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 — under the + # thresholds the production resolver actually produces. + events: list[tuple[str, dict[str, Any]]] = [] + plan = resolve_compaction_plan(window=200_000, output_limit=64_000, tool_buffer=15_000) + assert plan.enabled and plan.trigger_tokens is not None and plan.target_tokens is not None + cfg = _make_config( + trigger_tokens=plan.trigger_tokens, + target_tokens=plan.target_tokens, + event_callback=lambda t, d: events.append((t, d)), + ) + capability = build_tiered_compaction(cfg) + # CJK text tokenizes near one token per character, so the primary + # ~4-chars-per-token heuristic undercounts the suffix by ~4x: the primary + # estimate stays below the trigger while the real request is past the + # known window. + 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}-" + "日本" * 2_098)]) + ) + request_context = _request_context_with_messages(messages) + + from pydantic_ai.models import ModelRequestParameters + + primary_before = await _estimate_context_tokens(messages, ModelRequestParameters()) + density_before = await _estimate_context_tokens_density(messages, ModelRequestParameters()) + # The scenario must isolate the window guard: primary below the trigger, + # density-calibrated estimate at or above the window. + assert primary_before <= plan.trigger_tokens < density_before + assert density_before >= cfg.window_tokens + + result = await capability.before_model_request(_make_run_context(), request_context) + + assert len(result.messages) < len(messages) + density_after = await _estimate_context_tokens_density( + list(result.messages), ModelRequestParameters() + ) + assert density_after < cfg.window_tokens, ( + "compaction must bring the density-calibrated estimate back under the known window" + ) + + start = [d for t, d in events if t == "agent_compaction_start"] + complete = [d for t, d in events if t == "agent_compaction_complete"] + assert len(start) == 1 and len(complete) == 1 + assert start[0]["trigger_reason"] == "window_guard" + assert start[0]["density_tokens"] == density_before + # tokens_before stays on the primary token scale — the density value is + # a gate input, never token telemetry. + assert start[0]["tokens_before"] == primary_before + # The summarizing tier has no model in this harness, so it degrades and + # the sliding-window fallback produces the compacted history. + assert complete[0]["degraded_tiers"] == ["summarizing"] + assert complete[0]["degraded_estimators"] == [] + assert complete[0]["still_over_window"] is False + + @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 density-calibrated estimate shows the request can exceed the + # known window. + events: list[tuple[str, dict[str, Any]]] = [] + 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=110_000, + event_callback=lambda t, d: events.append((t, d)), + ), + ) + # A single token-dense message: the independent fallback measures ~1 token + # per CJK character and trips the window guard without the primary estimate. + request_context = _request_context_with_messages( + [ModelRequest(parts=[UserPromptPart(content="日本" * 100_001)])] + ) + + 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 + start = [d for t, d in events if t == "agent_compaction_start"] + complete = [d for t, d in events if t == "agent_compaction_complete"] + assert len(start) == 1 and len(complete) == 1 + assert start[0]["trigger_reason"] == "window_guard" + assert start[0]["tokens_before"] == 200_002 + assert complete[0]["degraded_estimators"] == ["primary"] + + @pytest.mark.asyncio + async def test_shared_harness_failure_still_compacts_via_independent_fallback(self) -> None: + # Requirement: a failure inside the shared harness estimator must not take + # the fallback down with it — the independent estimate walks the messages + # directly, so a real estimator bug still compacts before a large request. + 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=110_000), + ) + request_context = _request_context_with_messages( + [ModelRequest(parts=[UserPromptPart(content="日本" * 100_001)])] + ) + + with patch( + "pydantic_ai_harness.compaction.estimate_context_tokens", + side_effect=RuntimeError("harness estimator bug"), + ): + result = await capability.before_model_request(_make_run_context(), request_context) + + inner.before_model_request.assert_called_once() + assert result is compacted + + @pytest.mark.asyncio + async def test_ordinary_prose_below_trigger_is_not_compacted(self) -> None: + # Requirement: ordinary prose well below the trigger must neither compact + # nor emit compaction events — the density-calibrated guard fires only on + # genuinely token-dense content, never on a history that is merely large. + events: list[tuple[str, dict[str, Any]]] = [] + plan = resolve_compaction_plan(window=200_000, output_limit=64_000, tool_buffer=15_000) + assert plan.enabled and plan.trigger_tokens is not None and plan.target_tokens is not None + cfg = _make_config( + trigger_tokens=plan.trigger_tokens, + target_tokens=plan.target_tokens, + event_callback=lambda t, d: events.append((t, d)), + ) + capability = build_tiered_compaction(cfg) + prose = "The quick brown fox jumps over the lazy dog and then writes a " + messages = [ + ModelRequest(parts=[UserPromptPart(content=(prose * 70)[:4_000])]) for _ in range(50) + ] # 200,000 characters ~= 50,000 real tokens + 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 not events + + @pytest.mark.asyncio + async def test_window_guard_fires_at_exact_window_boundary(self) -> None: + # Requirement: the guard comparison is inclusive — a density estimate + # exactly equal to the known window must trip it. + events: list[tuple[str, dict[str, Any]]] = [] + cfg = _make_config( + trigger_tokens=121_000, + target_tokens=110_000, + event_callback=lambda t, d: events.append((t, d)), + ) + capability = build_tiered_compaction(cfg) + exact = _request_context_with_messages( + [ModelRequest(parts=[UserPromptPart(content="日本" * 100_000)])] + ) # density estimate == window_tokens exactly + + result = await capability.before_model_request(_make_run_context(), exact) + + start = [d for t, d in events if t == "agent_compaction_start"] + assert len(start) == 1 + assert start[0]["trigger_reason"] == "window_guard" + assert start[0]["density_tokens"] == 200_000 + assert len(result.messages) <= 1 + + events.clear() + below = _request_context_with_messages( + [ModelRequest(parts=[UserPromptPart(content="日" * 199_999)])] + ) + result_below = await capability.before_model_request(_make_run_context(), below) + assert len(result_below.messages) == 1 + assert not events + + +class TestDensityTextTokenBound: + """Requirement: the density bound escalates only for genuinely dense text.""" + + def test_cjk_text_counts_one_token_per_character(self) -> None: + # CJK tokenizes near one token per character, which the ~4-chars-per-token + # heuristic undercounts by ~4x — the case the window guard exists for. + assert _density_text_token_bound("日本語" * 1_000) == 3_000 + + def test_whitespace_poor_ascii_counts_two_chars_per_token(self) -> None: + # base64/hex/minified blobs tokenize near 1.5-2 characters per token. + assert _density_text_token_bound("a" * 10_000) == 5_001 + + def test_ordinary_prose_matches_primary_heuristic(self) -> None: + # Ordinary text must not be escalated: the bound stays the chars/4 + # heuristic so the guard never fires on a history that is merely large. + prose = "The quick brown fox jumps over the lazy dog and then writes a " + assert _density_text_token_bound(prose * 70) == len(prose * 70) // 4 + + def test_empty_text_is_zero(self) -> None: + assert _density_text_token_bound("") == 0 + + @pytest.mark.asyncio + async def test_density_estimate_counts_cjk_characters_not_heuristic(self) -> None: + # The harness-anchored density estimate must apply the density bound to + # message text, not the ~4-chars heuristic. + from pydantic_ai.models import ModelRequestParameters + + messages = [ModelRequest(parts=[UserPromptPart(content="日本語" * 1_000)])] + assert await _estimate_context_tokens_density(messages, ModelRequestParameters()) == 3_000 + + +class TestEstimatorFailureBranches: + """Requirement: estimator degradations are visible and never latch.""" + + @pytest.mark.asyncio + async def test_double_estimator_failure_emits_skipped_event_without_latch(self) -> None: + # Requirement: when both the primary and the fallback measurement fail, + # the request context is returned unchanged with an + # agent_compaction_skipped event and the disable latch stays off. + events: list[tuple[str, dict[str, Any]]] = [] + cfg = _make_config( + trigger_tokens=10, + target_tokens=5, + event_callback=lambda t, d: events.append((t, d)), + ) + capability = build_tiered_compaction(cfg) + request_context = _request_context_with_messages( + [ModelRequest(parts=[UserPromptPart(content="x" * 80_000)])] + ) + with ( + patch( + "conductor.providers._pydantic_ai.compaction._estimate_context_tokens", + new=AsyncMock(side_effect=RuntimeError("primary exploded")), + ), + patch( + "conductor.providers._pydantic_ai.compaction._estimate_context_tokens_independent", + new=AsyncMock(side_effect=RuntimeError("fallback exploded")), + ), + ): + result = await capability.before_model_request(_make_run_context(), request_context) + + assert result is request_context + assert capability._disabled is False # type: ignore[attr-defined] + assert [t for t, _ in events] == ["agent_compaction_skipped"] + assert events[0][1]["reason"] == "estimate_unavailable" + + @pytest.mark.asyncio + async def test_density_failure_compacts_on_primary_and_reports_degradation(self) -> None: + # Requirement: when the density-calibrated measurement fails but the + # primary succeeds, compaction proceeds on the primary alone, the + # complete event names the lost measurement in degraded_estimators, + # and the latch stays off. + events: list[tuple[str, dict[str, Any]]] = [] + cfg = _make_config( + trigger_tokens=10, + target_tokens=5, + event_callback=lambda t, d: events.append((t, d)), + ) + capability = build_tiered_compaction(cfg) + messages: list[Any] = [] + for i in range(40): + messages.append(ModelRequest(parts=[UserPromptPart(content=f"old {i:03d}")])) + messages.append(ModelRequest(parts=[UserPromptPart(content="x" * 80_000)])) + request_context = _request_context_with_messages(messages) + with patch( + "conductor.providers._pydantic_ai.compaction._estimate_context_tokens_density", + new=AsyncMock(side_effect=RuntimeError("density exploded")), + ): + result = await capability.before_model_request(_make_run_context(), request_context) + + assert len(result.messages) < len(messages) + complete = [d for t, d in events if t == "agent_compaction_complete"] + assert len(complete) == 1 + assert complete[0]["errored"] is False + assert complete[0]["degraded_estimators"] == ["density"] + assert capability._disabled is False # type: ignore[attr-defined] + + class TestTierFallback: """Requirement: a failing non-final tier still yields to the final tier.""" @@ -697,6 +989,8 @@ def callback(event_type: str, data: dict[str, Any]) -> None: "target_tokens", "messages_before", "tokens_before", + "trigger_reason", + "density_tokens", } assert set(start.keys()) == expected_start_keys assert start["agent_name"] == cfg.agent_name @@ -710,6 +1004,9 @@ def callback(event_type: str, data: dict[str, Any]) -> None: assert start["target_tokens"] == cfg.target_tokens assert start["messages_before"] == len(messages) assert start["tokens_before"] > cfg.trigger_tokens + assert start["trigger_reason"] == "trigger" + assert isinstance(start["density_tokens"], int) + assert start["density_tokens"] < cfg.window_tokens complete = events[types.index("agent_compaction_complete")][1] expected_complete_keys = { @@ -727,12 +1024,16 @@ def callback(event_type: str, data: dict[str, Any]) -> None: "errored", "degraded_tiers", "still_over_trigger", + "degraded_estimators", + "still_over_window", } assert set(complete.keys()) == expected_complete_keys assert complete["errored"] is False # The summarizing tier has no model outside a real run, so it degrades # and the sliding-window fallback produces the compacted history. assert complete["degraded_tiers"] == ["summarizing"] + assert complete["degraded_estimators"] == [] + assert complete["still_over_window"] is False assert complete["agent_name"] == cfg.agent_name assert complete["strategy"] == "tiered" assert complete["model"] == cfg.model_name @@ -920,9 +1221,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 density-calibrated + # 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 +1243,18 @@ 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", + ] + start = events[0][1] + complete = events[1][1] + # The whitespace-poor payload measures ~2 chars/token under the density + # bound, which drives the trigger when the primary estimate is lost. + assert start["tokens_before"] == 40_001 + assert start["density_tokens"] == 40_001 + assert start["trigger_reason"] == "trigger" + assert complete["degraded_estimators"] == ["primary"] @pytest.mark.asyncio async def test_telemetry_failure_keeps_compacted_result_without_latch(self) -> None: