ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

perf(compression): add guarded fast summary lane

Browse Source
This commit is contained in:
Mike DeMott
2026-08-23 15:00:01 -07:00
committed by kshitij
parent 01740f352e
commit 213ae08e7a
11 changed files with 362 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/auxiliary_client.py
+102 -20
View File
@@ -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)
agent/context_compressor.py
+30 -1
View File
@@ -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
agent/conversation_compression.py
+7
View File
@@ -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:
apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx
+13
View File
@@ -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({})
})
})
apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts
+2 -2
View File
@@ -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)) {
apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/session-info.ts
+9
View File
@@ -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
apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts
+3 -3
View File
@@ -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)
}
apps/desktop/src/store/compaction.ts
+10
View File
@@ -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)
}
}
hermes_cli/config_defaults.py
+4
View File
@@ -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
tests/agent/test_compression_attempt_telemetry.py
+49
View File
@@ -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)
tests/agent/test_fast_compression_lane.py
+133
View File
@@ -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"}