From 213ae08e7a8fead2d3f2525d178dcb5b32e39c46 Mon Sep 17 00:00:00 2001 From: Mike DeMott Date: Sun, 23 Aug 2026 15:00:01 -0700 Subject: [PATCH] perf(compression): add guarded fast summary lane --- agent/auxiliary_client.py | 122 +++++++++++++--- agent/context_compressor.py | 31 +++- agent/conversation_compression.py | 7 + .../compaction-event.test.tsx | 13 ++ .../use-message-stream/gateway-event/index.ts | 4 +- .../gateway-event/session-info.ts | 9 ++ .../gateway-event/status.ts | 6 +- apps/desktop/src/store/compaction.ts | 10 ++ hermes_cli/config_defaults.py | 4 + .../test_compression_attempt_telemetry.py | 49 +++++++ tests/agent/test_fast_compression_lane.py | 133 ++++++++++++++++++ 11 files changed, 362 insertions(+), 26 deletions(-) create mode 100644 tests/agent/test_fast_compression_lane.py diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index e371b0f676..0053c1cade 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -8359,6 +8359,41 @@ def _get_auxiliary_task_config(task: str) -> Dict[str, Any]: return task_config +class CompressionFastLane(NamedTuple): + """Explicit, non-reasoning compression route safe for a bounded summary.""" + + certified_non_reasoning: bool + max_tokens: Optional[int] + reasoning_config: Optional[Dict[str, Any]] + + +def resolve_compression_fast_lane() -> CompressionFastLane: + """Resolve the opt-in compression-only fast lane without changing defaults. + + A cap is safe only when the operator has selected a concrete auxiliary + provider/model and explicitly certified the route as non-reasoning. Auto + and inherited routes deliberately retain the historic uncapped request. + """ + config = _get_auxiliary_task_config("compression") + provider = str(config.get("provider") or "").strip().lower() + model = str(config.get("model") or "").strip() + effort = str(config.get("reasoning_effort") or "").strip().lower() + explicit_route = provider not in {"", "auto"} and model.lower() not in {"", "auto"} + certified = explicit_route and effort == "none" + 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, + {"enabled": False, "effort": "none"}, + ) + + def _get_task_timeout(task: str, default: float = _DEFAULT_AUX_TIMEOUT) -> float: """Read timeout from auxiliary.{task}.timeout in config, falling back to *default*.""" if not task: @@ -8413,6 +8448,18 @@ def _get_task_extra_body(task: str) -> Dict[str, Any]: result = dict(raw) if isinstance(raw, dict) else {} if "reasoning" not in result: effort = task_config.get("reasoning_effort") + if task == "compression" and str(effort or "").strip().lower() == "none": + # Do not let an inherited/main route silently inherit an explicit + # ``none`` intended for a configured fast summarizer. The main + # chat model and its reasoning policy remain untouched. + fast_lane = resolve_compression_fast_lane() + if fast_lane.reasoning_config is not None: + result["reasoning"] = fast_lane.reasoning_config + return result + # Other explicit compression reasoning policies retain their + # established behavior. This guard is narrowly about a claimed + # fast non-reasoning route. + return result if effort is not None and effort != "": if task in ("moa_reference", "moa_aggregator"): logger.warning( @@ -9380,38 +9427,61 @@ def call_llm( stream: bool = False, stream_options: dict = None, route_info: Optional[Dict[str, str]] = None, + latency_info: Optional[Dict[str, int]] = None, ) -> Any: """Run an auxiliary LLM request, applying the configured task limit.""" + queue_started_at = time.monotonic() semaphore = _acquire_sync_aux_semaphore(task) if semaphore is not None: semaphore.acquire() - try: - response = _call_llm_impl( - task=task, - provider=provider, - model=model, - base_url=base_url, - api_key=api_key, - main_runtime=main_runtime, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - tools=tools, - timeout=timeout, - extra_body=extra_body, - reasoning_config=reasoning_config, - extra_headers=extra_headers, - api_mode=api_mode, - stream=stream, - stream_options=stream_options, - route_info=route_info, + request_started_at = time.monotonic() + if latency_info is not None: + latency_info["queue_wait_ms"] = max( + 0, int((request_started_at - queue_started_at) * 1000) ) + + prior_progress_hook = getattr(_aux_progress, "hook", None) + + def _timed_progress() -> None: + if latency_info is not None and "time_to_first_progress_ms" not in latency_info: + latency_info["time_to_first_progress_ms"] = max( + 0, int((time.monotonic() - request_started_at) * 1000) + ) + if callable(prior_progress_hook): + prior_progress_hook() + + try: + with aux_progress_hook(_timed_progress if latency_info is not None else prior_progress_hook): + response = _call_llm_impl( + task=task, + provider=provider, + model=model, + base_url=base_url, + api_key=api_key, + main_runtime=main_runtime, + messages=messages, + temperature=temperature, + max_tokens=max_tokens, + tools=tools, + timeout=timeout, + extra_body=extra_body, + reasoning_config=reasoning_config, + extra_headers=extra_headers, + api_mode=api_mode, + stream=stream, + stream_options=stream_options, + route_info=route_info, + ) if stream and semaphore is not None: stream_semaphore = semaphore semaphore = None return _release_sync_semaphore_after_stream(response, stream_semaphore) return response finally: + if latency_info is not None: + latency_info["summary_generation_ms"] = max( + 0, int((time.monotonic() - request_started_at) * 1000) + ) if semaphore is not None: semaphore.release() @@ -9583,6 +9653,12 @@ def _call_llm_impl( f"Run: hermes setup") effective_timeout = _effective_aux_timeout(task, timeout) + fast_compression_lane = resolve_compression_fast_lane() if task == "compression" else None + force_fast_compression_cap = bool( + fast_compression_lane is not None + and fast_compression_lane.max_tokens is not None + and max_tokens == fast_compression_lane.max_tokens + ) request_provider = effective_provider or resolved_provider _set_relay_auxiliary_route( request_provider, @@ -9609,6 +9685,12 @@ def _call_llm_impl( tools=tools, timeout=effective_timeout, extra_body=effective_extra_body, reasoning_config=reasoning_config, base_url=_base_info or resolved_base_url, task=task) + if force_fast_compression_cap: + # Normal auxiliary calls intentionally omit a cap on most + # OpenAI-compatible/local providers. This is the narrow exception: + # the configured compression route is concrete and certified + # non-reasoning, so a bounded summary request is intentional. + kwargs.update(auxiliary_max_tokens_param(max_tokens, model=final_model)) if extra_headers: kwargs["extra_headers"] = dict(extra_headers) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index e98fbb6a08..62e08d31d1 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -33,6 +33,7 @@ from agent.auxiliary_client import ( _is_connection_error, aux_interrupt_protection, call_llm, + resolve_compression_fast_lane, ) from agent.context_engine import ContextEngine, sanitize_memory_context from agent.error_classifier import FailoverReason, classify_api_error @@ -2251,6 +2252,11 @@ class ContextCompressor(ContextEngine): "chunk_count": 0, "total_duration_ms": None, "aux_call_duration_ms": None, + "queue_wait_ms": None, + "prompt_build_ms": None, + "time_to_first_progress_ms": None, + "summary_generation_ms": None, + "commit_ms": None, "fallback_used": False, "commit_status": "unknown", "split_status": "unknown", @@ -2283,6 +2289,7 @@ class ContextCompressor(ContextEngine): aux_provider: str | None = None, aux_model: str | None = None, effective_aux_context: int | None = None, + phase_timings: Dict[str, Any] | None = None, ) -> None: telemetry = getattr(self, "_active_compression_telemetry", None) if not isinstance(telemetry, dict): @@ -2306,6 +2313,19 @@ class ContextCompressor(ContextEngine): ) previous = telemetry.get("aux_call_duration_ms") or 0 telemetry["aux_call_duration_ms"] = previous + max(0, int(duration_ms)) + for key in ( + "queue_wait_ms", + "prompt_build_ms", + "time_to_first_progress_ms", + "summary_generation_ms", + "commit_ms", + ): + if isinstance(phase_timings, dict) and key in phase_timings: + value = _safe_int(phase_timings[key]) + if key in {"queue_wait_ms", "summary_generation_ms"} and value is not None: + telemetry[key] = (telemetry.get(key) or 0) + value + else: + telemetry[key] = value def _emit_init_summary_once(self) -> None: """Emit the informative startup line once, on first resolution. @@ -4770,7 +4790,8 @@ Summary generation was unavailable, so this is a best-effort deterministic fallb the middle turns without a summary rather than inject a useless placeholder. """ - now = time.monotonic() + prompt_started_at = time.monotonic() + now = prompt_started_at if now < self._summary_failure_cooldown_until: logger.debug( "Skipping context summary during cooldown (%.0fs remaining)", @@ -5055,6 +5076,7 @@ FOCUS TOPIC: "{focus_topic}" This compaction should PRIORITISE preserving all information related to the focus topic above. For content related to "{focus_topic}", include full detail — exact values, file paths, command outputs, error messages, and decisions. For content NOT related to the focus topic, summarise more aggressively (brief one-liners or omit if truly irrelevant). The focus topic sections should receive roughly 60-70% of the summary token budget. Even for the focus topic, NEVER preserve API keys, tokens, passwords, or credentials — use [REDACTED].""" try: + fast_lane = resolve_compression_fast_lane() call_kwargs = { "task": "compression", "main_runtime": { @@ -5076,6 +5098,8 @@ This compaction should PRIORITISE preserving all information related to the focu # fall back to the model's native output ceiling. # timeout resolved from auxiliary.compression.timeout config by call_llm } + if fast_lane.max_tokens is not None: + call_kwargs["max_tokens"] = fast_lane.max_tokens if self.summary_model: call_kwargs["model"] = self.summary_model _aux_provider = "" @@ -5110,6 +5134,10 @@ This compaction should PRIORITISE preserving all information related to the focu # marker, losing the real handoff (#23975). Re-entrant: a main-model # retry (_generate_summary recursion) re-enters harmlessly. _aux_call_start = time.monotonic() + _latency_info: Dict[str, int] = { + "prompt_build_ms": max(0, int((_aux_call_start - prompt_started_at) * 1000)) + } + call_kwargs["latency_info"] = _latency_info try: with aux_interrupt_protection(): response = call_llm(**call_kwargs) @@ -5124,6 +5152,7 @@ This compaction should PRIORITISE preserving all information related to the focu aux_provider=_aux_provider, aux_model=_aux_model, effective_aux_context=_aux_context, + phase_timings=_latency_info, ) # ``_validate_llm_response`` only guarantees ``choices[0].message`` # exists, not that it's an object with ``.content``. Some diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index c9a0979f48..ee27095cd3 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -1393,6 +1393,7 @@ def _emit_compression_attempt_telemetry( commit_status: str, split_status: str, failure_class: str | None = None, + commit_started_at: float | None = None, ) -> None: """Emit one content-free JSON log line for a compression attempt.""" try: @@ -1406,6 +1407,10 @@ def _emit_compression_attempt_telemetry( payload["total_duration_ms"] = int((time.monotonic() - started_at) * 1000) payload["commit_status"] = commit_status payload["split_status"] = split_status + if commit_started_at is not None: + commit_ms = max(0, int((time.monotonic() - commit_started_at) * 1000)) + telemetry["commit_ms"] = commit_ms + payload["commit_ms"] = commit_ms if failure_class: payload["failure_class"] = failure_class payload.setdefault("chunking", False) @@ -3742,6 +3747,7 @@ def compress_context( agent._cached_system_prompt = new_system_prompt _session_commit_succeeded = False + _commit_started_at = time.monotonic() split_status = "not_applicable" if agent._session_db: split_status = "pending" @@ -4483,6 +4489,7 @@ def compress_context( if split_status in {"failed_not_indexed", "aborted"} else None ), + commit_started_at=_commit_started_at, ) return compressed, new_system_prompt finally: diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx index 0343dd37e3..7152d9eba5 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx @@ -55,4 +55,17 @@ describe('useMessageStream compaction lifecycle', () => { expect($compactingSessions.get()).toEqual({ [OTHER_SID]: true }) }) + + it('reconciles a reconnecting compaction only from trusted terminal server state', () => { + mountStream() + emit('status.update', { kind: 'compacting' }) + + // A running heartbeat is not terminal evidence and must not hide real work. + emit('session.info', { running: true }) + expect($compactingSessions.get()).toEqual({ [SID]: true }) + + // A server-reported terminal turn is trusted reconnect evidence. + emit('session.info', { running: false }) + expect($compactingSessions.get()).toEqual({}) + }) }) diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts index 693a03696e..df11da34be 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts @@ -7,7 +7,7 @@ import { resolveGatewayEventSessionId, UNSCOPED_STREAM_EVENT_TYPES } from '@/lib/gateway-events' -import { setSessionCompacting } from '@/store/compaction' +import { reconcileSessionCompacting } from '@/store/compaction' import { $gateway, activeGatewayConnectionId } from '@/store/gateway' import { $activeGatewayProfile, normalizeProfileKey } from '@/store/profile' import { replayPendingApproval } from '@/store/prompts' @@ -206,7 +206,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { // turn has resumed, so retire the phase label without waiting for the // whole turn to complete. if (sessionId && COMPACTION_RESUME_EVENT_TYPES.has(event.type) && compactedTurnRef.current.has(sessionId)) { - setSessionCompacting(sessionId, false) + reconcileSessionCompacting(sessionId, 'resumed') } if (sessionId && DRAFT_SUPERSEDING_EVENT_TYPES.has(event.type)) { diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/session-info.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/session-info.ts index 3e39bb872b..cdc713656a 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/session-info.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/session-info.ts @@ -1,6 +1,7 @@ import { normalizePersonalityValue } from '@/lib/chat-runtime' import { modelOptionsQueryKey } from '@/lib/model-options' import { reconcileApprovalModeForProfile } from '@/store/approval-mode' +import { reconcileSessionCompacting } from '@/store/compaction' import { requestDesktopOnboardingForCredentialWarning } from '@/store/onboarding' import { followActiveSessionCwd } from '@/store/projects' import { @@ -140,6 +141,14 @@ export function handleSessionInfoEvent(ctx: GatewayEventContext): boolean { const modelChanged = typeof payload?.model === 'string' const providerChanged = typeof payload?.provider === 'string' const runningChanged = typeof payload?.running === 'boolean' + + // Reconnect can miss the structured `compacted` edge. A gateway-authored + // running=false is a terminal fact; a running heartbeat is intentionally + // not used as a timeout-like guess, so genuine compaction stays visible. + if (sessionId && payload?.running === false) { + reconcileSessionCompacting(sessionId, 'terminal') + } + // The backend stamps model/provider (as strings) on EVERY session.info, // so the presence flags above are true on every heartbeat/turn edge — // fine for the cheap atom writes below (nanostores skips identical diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts index 37f56cbc68..3b5956f1ab 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts @@ -4,7 +4,7 @@ import { coerceGatewayText } from '@/lib/chat-runtime' import { isProviderSetupErrorMessage } from '@/lib/provider-setup-errors' import { type AgentNoticePayload, clearAgentNotice, nativeNoticeInput, showAgentNotice } from '@/store/agent-notices' import { clearClarifyRequest } from '@/store/clarify' -import { setSessionCompacting } from '@/store/compaction' +import { reconcileSessionCompacting, setSessionCompacting } from '@/store/compaction' import { refreshBackgroundProcesses } from '@/store/composer-status' import { applyGoalStatusText } from '@/store/goals' import { dispatchNativeNotification } from '@/store/native-notifications' @@ -28,7 +28,7 @@ export function handleStatusEvent(ctx: GatewayEventContext): boolean { setSessionCompacting(sessionId, true) compactedTurnRef.current.add(sessionId) } else if (sessionId && payload?.kind === 'compacted') { - setSessionCompacting(sessionId, false) + reconcileSessionCompacting(sessionId, 'terminal') compactedTurnRef.current.delete(sessionId) } else if (sessionId && payload?.kind === 'process') { // The gateway's notification poller announces background process @@ -128,7 +128,7 @@ export function handleStatusEvent(ctx: GatewayEventContext): boolean { clearAllPrompts(sessionId) clearClarifyRequest(undefined, sessionId) clearActiveSessionTodos(sessionId) - setSessionCompacting(sessionId, false) + reconcileSessionCompacting(sessionId, 'terminal') compactedTurnRef.current.delete(sessionId) } diff --git a/apps/desktop/src/store/compaction.ts b/apps/desktop/src/store/compaction.ts index cbe067ee3a..97a583882e 100644 --- a/apps/desktop/src/store/compaction.ts +++ b/apps/desktop/src/store/compaction.ts @@ -35,3 +35,13 @@ export function setSessionCompacting(sessionId: string | null | undefined, activ delete next[key] $compactingSessions.set(next) } + +/** Clear compaction only when the gateway proves the turn resumed or ended. */ +export function reconcileSessionCompacting( + sessionId: string | null | undefined, + evidence: 'resumed' | 'terminal' +): void { + if (evidence === 'resumed' || evidence === 'terminal') { + setSessionCompacting(sessionId, false) + } +} diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 59a9e2d979..6841c3e553 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1135,6 +1135,10 @@ DEFAULT_CONFIG = { "timeout": 120, # seconds — compression summarises large contexts; increase for local models "extra_body": {}, "reasoning_effort": "", # per-task thinking level: none|minimal|low|medium|high|xhigh|max|ultra (empty = provider default) + # Guarded fast lane: only honored with a concrete provider/model + # and an explicit ``reasoning_effort: none`` certification. + # Zero preserves the historic uncapped compression request. + "max_output_tokens": 0, }, # Note: session_search no longer uses an auxiliary LLM (PR #27590 — # single-shape tool returns DB content directly). The old diff --git a/tests/agent/test_compression_attempt_telemetry.py b/tests/agent/test_compression_attempt_telemetry.py index 9b2b587d29..5035eef1d0 100644 --- a/tests/agent/test_compression_attempt_telemetry.py +++ b/tests/agent/test_compression_attempt_telemetry.py @@ -104,6 +104,11 @@ def test_compression_attempt_telemetry_is_metadata_only(caplog): assert payload["split_status"] == "not_applicable" assert payload["fallback_used"] is False assert isinstance(payload["total_duration_ms"], int) + assert isinstance(payload["commit_ms"], int) + assert payload["queue_wait_ms"] is None + assert payload["prompt_build_ms"] is None + assert payload["time_to_first_progress_ms"] is None + assert payload["summary_generation_ms"] is None raw_log = json.dumps(payload) assert "TOPSECRET_TRANSCRIPT_TEXT" not in raw_log @@ -149,3 +154,47 @@ def test_aux_call_telemetry_records_durations_without_content(caplog): raw_log = json.dumps(payload) assert "TOPSECRET_TRANSCRIPT_TEXT" not in raw_log assert "SANITIZED SUMMARY" not in raw_log + + +def test_aux_call_telemetry_records_content_free_phase_timings(): + with patch("agent.context_compressor.get_model_context_length", return_value=100_000): + compressor = ContextCompressor( + model="test/main-model", + provider="test-provider", + threshold_percent=0.50, + quiet_mode=True, + config_context_length=100_000, + ) + compressor._begin_compression_telemetry(current_tokens=75_000) + + compressor._record_aux_compression_call( + prompt_messages=[{"role": "user", "content": "TOPSECRET_TRANSCRIPT_TEXT"}], + max_tokens=1400, + duration_ms=22, + aux_provider="ollama", + aux_model="qwen3:8b", + phase_timings={ + "queue_wait_ms": 3, + "prompt_build_ms": 5, + "time_to_first_progress_ms": 7, + "summary_generation_ms": 19, + "commit_ms": 11, + }, + ) + + payload = compressor._last_compression_telemetry + assert payload is not None + assert {key: payload[key] for key in ( + "queue_wait_ms", + "prompt_build_ms", + "time_to_first_progress_ms", + "summary_generation_ms", + "commit_ms", + )} == { + "queue_wait_ms": 3, + "prompt_build_ms": 5, + "time_to_first_progress_ms": 7, + "summary_generation_ms": 19, + "commit_ms": 11, + } + assert "TOPSECRET_TRANSCRIPT_TEXT" not in json.dumps(payload) diff --git a/tests/agent/test_fast_compression_lane.py b/tests/agent/test_fast_compression_lane.py new file mode 100644 index 0000000000..b8198bcdb6 --- /dev/null +++ b/tests/agent/test_fast_compression_lane.py @@ -0,0 +1,133 @@ +"""Contract tests for the opt-in non-reasoning compression fast lane.""" + +import time +from unittest.mock import MagicMock, patch + + +def _resolve(config): + from agent.auxiliary_client import resolve_compression_fast_lane + + with patch( + "agent.auxiliary_client._get_auxiliary_task_config", + return_value=config, + ): + return resolve_compression_fast_lane() + + +def test_explicit_non_reasoning_compression_route_is_certified_and_bounded(): + lane = _resolve( + { + "provider": "ollama", + "model": "qwen3:8b", + "reasoning_effort": "none", + "max_output_tokens": 1400, + } + ) + + assert lane.certified_non_reasoning is True + assert lane.max_tokens == 1400 + assert lane.reasoning_config == {"enabled": False, "effort": "none"} + + +def test_inherited_auto_or_uncertified_compression_routes_remain_uncapped(): + inherited = _resolve({"provider": "auto", "model": "", "reasoning_effort": "none", "max_output_tokens": 1400}) + unknown = _resolve({"provider": "ollama", "model": "qwen3:8b", "max_output_tokens": 1400}) + reasoning = _resolve( + { + "provider": "ollama", + "model": "qwen3:8b", + "reasoning_effort": "low", + "max_output_tokens": 1400, + } + ) + + for lane in (inherited, unknown, reasoning): + assert lane.certified_non_reasoning is False + assert lane.max_tokens is None + assert lane.reasoning_config is None + + +def test_only_a_certified_lane_forwards_non_reasoning_request_controls(): + from agent.auxiliary_client import _get_task_extra_body + + certified = { + "provider": "ollama", + "model": "qwen3:8b", + "reasoning_effort": "none", + "max_output_tokens": 1400, + } + inherited = { + "provider": "auto", + "model": "", + "reasoning_effort": "none", + "max_output_tokens": 1400, + } + + with patch("agent.auxiliary_client._get_auxiliary_task_config", return_value=certified): + assert _get_task_extra_body("compression")["reasoning"] == { + "enabled": False, + "effort": "none", + } + with patch("agent.auxiliary_client._get_auxiliary_task_config", return_value=inherited): + assert "reasoning" not in _get_task_extra_body("compression") + + +def test_compression_latency_records_queue_wait_and_first_progress(): + from agent.auxiliary_client import _notify_aux_progress, call_llm + + class _DelayedSemaphore: + def acquire(self): + time.sleep(0.01) + + def release(self): + pass + + timings = {} + + def _progressing_call(**_kwargs): + _notify_aux_progress() + return object() + + with ( + patch("agent.auxiliary_client._acquire_sync_aux_semaphore", return_value=_DelayedSemaphore()), + patch("agent.auxiliary_client._call_llm_impl", side_effect=_progressing_call), + ): + call_llm( + task="compression", + messages=[{"role": "user", "content": "summary request"}], + latency_info=timings, + ) + + assert timings["queue_wait_ms"] >= 5 + assert timings["time_to_first_progress_ms"] >= 0 + assert timings["summary_generation_ms"] >= timings["time_to_first_progress_ms"] + + +def test_certified_fast_lane_sends_the_configured_cap_to_its_provider(): + from agent.auxiliary_client import call_llm + + config = { + "provider": "ollama", + "model": "qwen3:8b", + "reasoning_effort": "none", + "max_output_tokens": 1400, + } + client = MagicMock() + client.base_url = "http://127.0.0.1:11434/v1" + response = object() + client.chat.completions.create.return_value = response + + with ( + patch("agent.auxiliary_client._get_auxiliary_task_config", return_value=config), + patch("agent.auxiliary_client._get_cached_client", return_value=(client, "qwen3:8b")), + patch("agent.auxiliary_client._validate_llm_response", return_value=response), + ): + assert call_llm( + task="compression", + messages=[{"role": "user", "content": "summary request"}], + max_tokens=1400, + ) is response + + request = client.chat.completions.create.call_args.kwargs + assert request["max_tokens"] == 1400 + assert request["extra_body"]["reasoning"] == {"enabled": False, "effort": "none"}