diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 5b6e44e..6bea702 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -40,3 +40,6 @@ jobs:
- name: Run mypy
run: mypy incontext
+
+ - name: Run mypy for tests
+ run: mypy tests --exclude tests/typing
diff --git a/README.md b/README.md
index 90612db..23e5fd8 100644
--- a/README.md
+++ b/README.md
@@ -1,51 +1,95 @@
ⓘ
+[](https://pepy.tech/project/incontext)
+[](https://pepy.tech/project/incontext)
+[](https://coveralls.io/github/pomponchik/incontext?branch=develop)
+[](https://github.com/boyter/scc/)
+[](https://hitsofcode.com/github/pomponchik/incontext/view?branch=develop)
[](https://github.com/pomponchik/incontext/actions/workflows/tests_and_coverage.yml)
[](https://github.com/pomponchik/incontext/actions/workflows/hermes_e2e.yml)
[](https://pypi.org/project/incontext/)
[](https://pypi.org/project/incontext/)
[](https://mypy-lang.org/)
[](https://github.com/astral-sh/ruff)
+[](https://deepwiki.com/pomponchik/incontext)
-`incontext` is a Hermes Agent plugin that gives each LLM request a viable
-output budget below Hermes' context-compression boundary. It prevents a fixed,
-oversized `max_tokens` value from consuming input space, but also refuses to
-manufacture tiny completions that are technically valid and operationally
-useless to an agent.
+`incontext` is a [Hermes Agent](https://github.com/NousResearch/hermes-agent)
+plugin that dynamically budgets output space against Hermes' active context
+boundary. It inserts a cap only when the remaining space meets the configured
+output reserve or a smaller caller-supplied cap; otherwise it leaves the request
+unchanged.
## Algorithm
-The policy uses these values:
-
-- `W` is Hermes' effective compression window, read from the installed
- `ContextCompressor`.
-- `P` is the exact number of tokens in the provider-visible prompt.
-- `R` is the minimum viable output reserve. It is configured with
- `INCONTEXT_MIN_OUTPUT_TOKENS` and defaults to `4096`. An explicit smaller
- Hermes-wide output cap lowers `R`, because that cap is an operator decision.
+On normal main turns with automatic compression enabled, the policy runs in two
+stages: preflight requests compression when needed, then middleware caps the
+output after the final request is built. With compression disabled, only the
+middleware stage runs, using `W` as defined below. The policy uses these values:
+
+- `W` is incontext's active budgeting boundary in tokens. By default, it is the
+ compressor threshold when automatic compression is enabled, or the full
+ context window when it is disabled. With compression enabled, the threshold
+ is resolved by the installed
+ [`ContextCompressor`](https://hermes-agent.nousresearch.com/docs/developer-guide/context-compression-and-caching/);
+ the emergency override supplies `W` directly but does not reconfigure Hermes.
+ With automatic compression, an override must match the active context
+ engine's actual boundary for preflight and middleware to share the same `W`.
+- `P` is the prompt count used for budgeting: an exact or conservative
+ provider-aware count when available, otherwise Hermes' rough estimate plus
+ `F`.
+- `R` is the configured output reserve. It is set with
+ `INCONTEXT_MIN_OUTPUT_TOKENS` and defaults to `4096`. If Hermes has a smaller
+ explicit output cap for the active route, that cap becomes `R`; the plugin
+ treats the operator's smaller limit as intentional.
- `B` is an optional positive output cap on an individual request.
-- `F` is `INCONTEXT_FALLBACK_MARGIN_TOKENS`, used only when exact tokenization
- is unavailable.
+- `F` is `INCONTEXT_FALLBACK_MARGIN_TOKENS`, used only when backend counting is
+ unavailable.
+
+```mermaid
+flowchart TD
+ A["Preflight counts P
(with the backend, or the rough estimate + F)"]
+ B{"P + R - 1 >= W?"}
+ C["Hermes evaluates the active engine's
compression policy and guards"]
+ D["Hermes builds the final request"]
+ E["Middleware recounts P
(with the backend, or the rough estimate + F)"]
+ F["Set required_output
to min(R, B), or R if B is absent"]
+ G{"remaining >= required_output?"}
+ H["Insert the calculated output cap"]
+ I["Leave the request unchanged
(fail open)"]
+
+ A --> B
+ B -- Yes --> C
+ B -- No --> C
+ C --> D
+ D --> E
+ E --> F
+ F --> G
+ G -- Yes --> H
+ G -- No --> I
+```
-Before Hermes constructs the main provider request, incontext reports this
-token pressure to its compression preflight:
+When automatic compression is enabled, incontext reports token pressure before
+Hermes constructs the main provider request. The individual request cap `B` is
+not known at this stage, so preflight uses `R`:
```text
preflight_pressure = P + R - 1
```
-Hermes compresses when that pressure is at least `W`. Therefore compression is
-requested exactly when `W - P < R`. The subtraction of one is intentional: a
-prompt with exactly `R` tokens of output space remains valid, while a prompt
-with `R - 1` tokens does not.
+incontext makes the reported pressure reach or exceed `W` exactly when
+`W - P < R`. Hermes then evaluates compression under its own guards, so reaching
+`W` does not guarantee that compression will run. The subtraction of one is
+intentional: a prompt with exactly `R` tokens of output space passes this check,
+while a prompt with `R - 1` tokens does not.
-After the final request has been constructed, the middleware counts it again
-and computes:
+After constructing the final request, the middleware recounts its
+provider-visible prompt, so `P` may differ from the preflight value, and
+computes:
```text
required_output = min(R, B) if B is present else R
@@ -54,35 +98,38 @@ max_tokens = min(remaining, B) if B is present else remaining
```
The middleware inserts the output cap only when `remaining >= required_output`.
-Otherwise it leaves the request unchanged: the preflight path owns main-turn
-compression, and fail-open behavior is safer for call sites that bypass it than
-sending a predictably truncated tool call or text fragment. A provider-specific
-output constraint must also leave at least `required_output` tokens.
-
-A smaller positive caller cap is preserved and becomes that request's required
-minimum. This keeps deliberately bounded operations, such as context summaries
-and generated titles, bounded. Without an explicit smaller cap, incontext never
+Otherwise it leaves the request unchanged instead of forcing a predictably
+truncated tool call or text fragment. With automatic compression enabled,
+preflight requests compression on normal main turns; the same fail-open behavior
+protects call sites that bypass it. Any additional wire-level output limit
+reported by the backend must also leave at least `required_output` tokens.
+
+If the caller supplies a positive cap below `R`, incontext preserves it and
+requires at least that much remaining space before inserting an output cap.
+This keeps deliberately bounded operations, such as context summaries and
+generated titles, bounded. Without such a caller cap, incontext never
dynamically emits `max_tokens` below `R`; in particular it does not turn an
exhausted window into `max_tokens=1`.
Hermes auxiliary calls do not pass through the public `llm_request` middleware,
and Hermes omits `max_tokens` for most custom providers. The plugin therefore
-wraps the auxiliary request builder with the same budgeting rule. If exact
-tokenization fails, both preflight and middleware use Hermes' rough estimate
-plus `F`, so they retain the same decision boundary. If both estimators fail,
-the original request is left unchanged.
+applies the same budgeting rule to auxiliary requests that use the configured
+primary route; requests to another model, provider, or endpoint pass through
+unchanged. If backend counting fails, both preflight and middleware use their
+respective Hermes rough estimates plus `F`. If both estimators fail in
+middleware, the original request is left unchanged.
-Startup rejects `F + R >= W`. Hermes normalizes a rough prompt estimate to at
-least one token, so that configuration could never leave `R` viable output
-tokens during a tokenizer outage, even after compressing everything else.
+Startup rejects `F + R >= W`, because that would leave no room for even the
+smallest fallback-counted prompt.
This addresses the same output-budget arithmetic discussed in
[NousResearch/hermes-agent#38652](https://github.com/NousResearch/hermes-agent/issues/38652).
## Installation
-Install the published package from PyPI and enable it using the same plugin
-name, `incontext`:
+Install the published package from [PyPI](https://pypi.org/project/incontext/)
+and [enable it](https://hermes-agent.nousresearch.com/docs/user-guide/features/plugins/)
+using the same plugin name, `incontext`:
```bash
python -m pip install incontext
@@ -96,16 +143,20 @@ python -m pip install 'git+https://github.com/pomponchik/incontext.git@develop'
hermes plugins enable incontext
```
-Restart the long-running Hermes gateway after installing or upgrading the
-Python package. Hermes discovers it through the official
-`hermes_agent.plugins` entry-point group; no source file has to be copied into
-`$HERMES_HOME/plugins`.
+After installing or upgrading, configure the plugin as described below and
+then restart any long-running Hermes gateway. Hermes discovers the plugin
+through the official `hermes_agent.plugins` entry-point group; no source file
+has to be copied into `$HERMES_HOME/plugins`.
## Configuration
-The bundled `vllm` backend is selected by default. Its
-`INCONTEXT_TOKENIZER_URL` setting is required and must point to the `/tokenize`
-endpoint of the same vLLM model Hermes uses:
+The bundled [vLLM](https://docs.vllm.ai/) backend is selected by default. Hermes
+must provide `model.default` and a positive `model.context_length`. For the
+bundled backend, `INCONTEXT_TOKENIZER_URL` is required and must point to a
+[`/tokenize` endpoint](https://docs.vllm.ai/en/stable/serving/online_serving/#tokenize-apis)
+with the same model, tokenizer, and chat-template configuration as Hermes'
+primary inference route. This example also shows the most commonly adjusted
+optional settings at their default values:
```bash
export INCONTEXT_BACKEND='vllm'
@@ -115,57 +166,49 @@ export INCONTEXT_FALLBACK_MARGIN_TOKENS='1024'
export INCONTEXT_MIN_OUTPUT_TOKENS='4096'
```
-Environment variables are loaded through typed `skelet.Storage` fields backed
-by ordered `skelet.EnvSource` instances. Primary `INCONTEXT_*` names take
-precedence over the supported legacy aliases. Text normalization and blank
-value rejection are implemented by the fields' native `conversion` and
-`validation` rules.
-
-Hermes' `model.default`, `model.context_length`, and `compression.threshold`
-remain the source of truth. The plugin constructs Hermes' installed
-`ContextCompressor` and uses its resolved `threshold_tokens`; it does not copy
-version-sensitive threshold arithmetic.
+Hermes' `model.default`, `model.context_length`, and `compression` settings
+remain the source of truth. With automatic compression enabled, the plugin
+constructs Hermes' installed `ContextCompressor` and uses its resolved
+`threshold_tokens` instead of copying version-sensitive arithmetic. With
+compression disabled, it uses `model.context_length`; the emergency override
+`INCONTEXT_COMPRESSION_WINDOW_TOKENS` bypasses this discovery and declares the
+budgeting boundary. It does not reconfigure Hermes' compressor or context
+engine.
-The optional variables are:
+The remaining variables are optional unless noted otherwise:
| Variable | Default | Meaning |
|---|---:|---|
-| `INCONTEXT_BACKEND` | `vllm` | Named `pristan` backend plugin |
+| `INCONTEXT_BACKEND` | `vllm` | Backend name registered in `incontext.backends`; `vllm` is bundled |
| `INCONTEXT_TOKENIZER_TIMEOUT_SECONDS` | `30` | `/tokenize` request timeout |
| `INCONTEXT_TOKENIZER_USER_AGENT` | automatic | HTTP user agent derived from installed package metadata |
-| `INCONTEXT_FALLBACK_MARGIN_TOKENS` | `1024` | Extra reserve only when exact tokenization fails |
-| `INCONTEXT_MIN_OUTPUT_TOKENS` | `4096` | Minimum viable output budget before compression is required |
-| `INCONTEXT_COMPRESSION_WINDOW_TOKENS` | unset | Explicit emergency override for the resolved Hermes boundary |
+| `INCONTEXT_FALLBACK_MARGIN_TOKENS` | `1024` | Extra reserve only when backend counting fails |
+| `INCONTEXT_MIN_OUTPUT_TOKENS` | `4096` | Base output reserve; smaller active-route and request caps are handled as described above |
+| `INCONTEXT_COMPRESSION_WINDOW_TOKENS` | unset | Explicit budgeting-boundary assertion; required with a non-default Hermes context engine |
The former `HERMES_VLLM_TOKENIZER_*` and
-`HERMES_DYNAMIC_BUDGET_FALLBACK_MARGIN_TOKENS` names are accepted as migration
-aliases. New deployments should use the `INCONTEXT_*` names.
+`HERMES_DYNAMIC_BUDGET_FALLBACK_MARGIN_TOKENS` names remain supported for
+migration, but `INCONTEXT_*` names take precedence and should be used in new
+deployments.
-## Replacing the inference backend
+## Replacing the budgeting backend
-The budgeting core depends only on the abstract `incontext.Backend` contract.
-It has no import or construction dependency on vLLM. A backend supplies its
-safe diagnostic `source`, exact `count(...)`, cache invalidation, and an
-optional output-field normalization hook.
+The budgeting core depends only on the abstract `incontext.Backend` contract,
+not on vLLM itself. A backend provides provider-aware prompt counting, cache
+invalidation, a non-sensitive name for logs (`source`), and optional
+normalization of provider-specific output fields.
-Backend implementations are named `pristan` plugins in the
-`incontext.backends` entry-point group. The generic `skelet` environment has a
-typed `backend` field whose default is `vllm`. At runtime incontext performs
-the single named resolution directly:
-
-```python
-backend = backends[environment.backend].one()
-```
+Backends are registered by name and discovered through
+[Python entry points](https://packaging.python.org/en/latest/specifications/entry-points/)
+in the `incontext.backends` group. `INCONTEXT_BACKEND` selects one and defaults
+to `vllm`.
-The `incontext` distribution itself publishes the `vllm` entry point. Loading
-that entry point imports `incontext.vllm_provider`, whose only responsibility is
-to construct `VllmBackend`. All `/tokenize` payload rules, vLLM response fields,
-context-length validation, transport settings, and caching live inside that
-class rather than in the budgeting core.
+The bundled `vllm` backend keeps all vLLM-specific tokenization and transport
+logic outside the budgeting core.
A third-party distribution can provide another backend without changing
-incontext. Its implementation subclasses the stable abstract contract and its
-plugin module registers a provider under a new name:
+incontext. Its implementation subclasses the abstract contract and its
+plugin module registers the backend under a new name:
```python
# acme_backend/plugin.py
@@ -205,39 +248,38 @@ The third-party package makes that module discoverable in `pyproject.toml`:
acme = "acme_backend.plugin"
```
-After installing the package, select it through the same typed configuration
-field and restart the Hermes process:
+After installing the package, set `INCONTEXT_BACKEND` to its registered name
+and restart Hermes:
```bash
export INCONTEXT_BACKEND='acme'
```
-Only the selected provider is instantiated. An unknown name fails `.one()`;
-the unique slot rejects duplicate providers under the same name while loading
-entry points. Startup therefore fails instead of choosing a backend implicitly.
-Each backend owns and validates its backend-specific configuration; the generic
-settings object contains only the compression-window, viability-reserve, and
-fallback-budget policy.
+Startup fails if the selected backend is missing or registered more than once.
+Each backend owns its specific settings.
## Safety properties
-- With the bundled backend, vLLM applies its real chat template to messages, tools, and
- `chat_template_kwargs`; local tokenizer approximations are not used.
-- The returned `max_model_len` must equal Hermes' configured context length.
-- `max_tokens`, `max_completion_tokens`, and `max_output_tokens` are reduced to
- the smallest positive caller cap while preserving the corresponding
- provider-selected field name; an implicit cap below the minimum viable
- output reserve is handed to preflight compression.
+- For ordinary chat requests, the bundled backend has vLLM's `/tokenize`
+ endpoint apply its real chat template to messages, tools, and
+ `chat_template_kwargs`; no local tokenizer approximation is used on that
+ path.
+- The `max_model_len` returned by vLLM's `/tokenize` endpoint must equal Hermes'
+ configured context length.
+- If a request contains several supported output-cap fields (`max_tokens`,
+ `max_completion_tokens`, or `max_output_tokens`), incontext uses the smallest
+ positive value and emits the field expected by the backend. If an additional
+ backend-reported limit is below the required reserve, the request is left
+ unchanged.
- The incoming request is copied and never mutated.
-- Exact counts use a bounded, thread-safe cache.
-- The exact counter is also used by Hermes' preflight compressor, eliminating
- the former gap where compression used a rough count but budgeting used an
- exact one.
+- The bundled counter uses a bounded, thread-safe cache.
+- The same counter is used for preflight and final budgeting.
- If `/tokenize` fails, Hermes' own rough estimator is used with an additional
safety margin. If both counters fail, the middleware leaves the request
unchanged instead of taking Hermes down.
-- Logs contain counts and exception types, never prompts, credentials, or raw
- provider errors.
+- With the bundled backend, incontext's own log messages contain counts and
+ exception types, never prompts, credentials, or raw provider errors.
-The tokenizer endpoint sees the prompt content by design. Run it on a trusted
-network path and use the same access controls as the inference endpoint.
+The tokenizer endpoint sees the prompt content by design. Keep `/tokenize` on a
+trusted network path and protect it with network-level controls; vLLM's built-in
+API-key check does not cover this route.
diff --git a/incontext/auxiliary.py b/incontext/auxiliary.py
index 306adcf..70cc106 100644
--- a/incontext/auxiliary.py
+++ b/incontext/auxiliary.py
@@ -45,18 +45,15 @@ def acquire(
output_cap_selector: Optional[OutputCapSelector],
) -> None:
"""Add one installation without mutating an earlier owner's state."""
-
self._owners = (*self._owners, (owner, runtime, output_cap_selector))
def release(self, owner: object) -> None:
"""Remove exactly one installation while retaining all other owners."""
-
self._owners = tuple(entry for entry in self._owners if entry[0] is not owner)
@property
def owned(self) -> bool:
"""Return whether at least one live installation owns this wrapper."""
-
return bool(self._owners)
def __call__(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
@@ -120,7 +117,6 @@ def _output_cap(
output_cap_selector: Optional[OutputCapSelector],
) -> Dict[str, int]:
"""Select Hermes' provider-specific output-cap alias safely."""
-
if output_cap_selector is not None:
try:
selected = output_cap_selector(value, model=model)
@@ -142,7 +138,6 @@ def _configuration(
self,
) -> Tuple[RuntimeSource, Optional[OutputCapSelector]]:
"""Read one coherent owner snapshot for the complete request."""
-
owners = self._owners
if owners:
_, runtime, output_cap_selector = owners[-1]
@@ -176,7 +171,6 @@ def _new_wrapper(
output_cap_selector: Optional[OutputCapSelector],
) -> Optional[_AuxiliaryBudget]:
"""Construct an auxiliary wrapper when its callable can be inspected."""
-
original = (
current.original
if isinstance(current, _AuxiliaryBudget)
@@ -194,7 +188,6 @@ def _new_wrapper(
def _load_auxiliary_builder() -> Optional[Tuple[Any, AuxiliaryBuilder]]:
"""Import and validate Hermes' optional private auxiliary binding."""
-
try:
auxiliary_client = import_module("agent.auxiliary_client")
except ImportError:
@@ -214,7 +207,6 @@ def _load_auxiliary_builder() -> Optional[Tuple[Any, AuxiliaryBuilder]]:
def install(runtime: RuntimeSource) -> Optional[Cleanup]:
"""Apply incontext to Hermes requests that bypass ``llm_request`` middleware."""
-
loaded = _load_auxiliary_builder()
if loaded is None:
return None
@@ -244,7 +236,6 @@ def install(runtime: RuntimeSource) -> Optional[Cleanup]:
def cleanup() -> None:
"""Release one owner and restore Hermes after the final unload."""
-
nonlocal closed
global _installed_wrapper # noqa: PLW0603
with _install_lock:
diff --git a/incontext/backend.py b/incontext/backend.py
index c55acd5..a5cf053 100644
--- a/incontext/backend.py
+++ b/incontext/backend.py
@@ -15,7 +15,6 @@ class Backend(ABC):
@abstractmethod
def source(self) -> str:
"""Return a stable, non-sensitive diagnostic source name."""
-
raise NotImplementedError
@abstractmethod
@@ -26,18 +25,15 @@ def count(
context_length: int,
) -> int:
"""Return the provider-visible prompt size in tokens."""
-
raise NotImplementedError
@abstractmethod
def clear_cache(self) -> None:
"""Discard backend-local cached data."""
-
raise NotImplementedError
def output_budget_field(self, requested_field: str) -> str:
"""Return the provider-supported wire alias for an output budget."""
-
return requested_field
def output_budget_limit(
@@ -47,13 +43,11 @@ def output_budget_limit(
context_length: int,
) -> Optional[int]:
"""Return an additional provider wire limit, when one exists."""
-
del request, context_length
return None
def coerce_output_budget(self, value: Any) -> Optional[int]:
"""Return a positive caller cap accepted by this provider."""
-
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
return None
return int(value)
@@ -66,5 +60,4 @@ def coerce_output_budget(self, value: Any) -> Optional[int]:
)
def backends() -> List[Backend]:
"""Provide named inference backends discovered through package metadata."""
-
return []
diff --git a/incontext/budget.py b/incontext/budget.py
index fb8d08a..0cb86b2 100644
--- a/incontext/budget.py
+++ b/incontext/budget.py
@@ -15,7 +15,6 @@
def _backend_source(backend: Backend) -> str:
"""Read optional diagnostics without letting them break middleware."""
-
try:
return backend.source
except Exception as backend_error: # noqa: BLE001
@@ -28,7 +27,6 @@ def _backend_source(backend: Backend) -> str:
def _materialize_request(request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Copy reusable OpenAI collections into their provider-visible shapes."""
-
messages = request.get("messages")
if not isinstance(messages, Collection) or isinstance(
messages,
@@ -64,7 +62,6 @@ def compute_max_tokens(
request while retaining ``1`` as the default for callers of this helper
that only need its original positive-space contract.
"""
-
if compression_window <= 0:
raise ValueError("compression_window must be positive")
if prompt_tokens < 0:
@@ -82,7 +79,6 @@ def compression_pressure_tokens(
minimum_output_tokens: int,
) -> int:
"""Include the viability reserve in Hermes' preflight pressure count."""
-
if prompt_tokens < 0:
raise ValueError("prompt_tokens must not be negative")
if minimum_output_tokens <= 0:
@@ -98,7 +94,6 @@ def _requested_output_cap(
coerce: Callable[[Any], Optional[int]],
) -> Optional[Tuple[str, int]]:
"""Return the provider field and smallest valid caller cap."""
-
extra_body = request.get("extra_body")
top_level_caps = [
(field, value)
@@ -123,7 +118,6 @@ def _requested_output_cap(
def estimate_request_tokens_rough(request: Dict[str, Any]) -> int:
"""Use Hermes' own conservative request estimator as a fallback."""
-
# Hermes is intentionally an optional runtime dependency of the PyPI package.
from agent.model_metadata import ( # type: ignore[import-not-found] # noqa: PLC0415
estimate_request_tokens_rough as estimator,
@@ -174,7 +168,6 @@ def __call__(
**context: Any,
) -> Optional[Dict[str, Any]]:
"""Rewrite output-cap aliases into one exact dynamic ``max_tokens``."""
-
if not isinstance(request, dict):
return None
prepared_request = _materialize_request(request)
@@ -261,7 +254,6 @@ def _resolve_output_budget(
source: str,
) -> Optional[Tuple[str, int]]:
"""Combine the window, caller cap, and backend wire constraint."""
-
try:
requested_output_cap = _requested_output_cap(
request,
diff --git a/incontext/hermes.py b/incontext/hermes.py
index a50e8c6..6499ce5 100644
--- a/incontext/hermes.py
+++ b/incontext/hermes.py
@@ -22,7 +22,6 @@
def build_runtime() -> DynamicOutputBudget:
"""Construct a fully validated runtime."""
-
environment = Environment()
settings = load_settings(environment=environment)
backend = backends[environment.backend].one()
@@ -31,7 +30,6 @@ def build_runtime() -> DynamicOutputBudget:
def get_runtime() -> DynamicOutputBudget:
"""Return the runtime scoped to the active Hermes profile/home."""
-
key = _runtime_key()
with _runtime_lock:
runtime = _runtimes.get(key)
@@ -48,7 +46,6 @@ def get_runtime() -> DynamicOutputBudget:
def _profile_cleanup(key: str) -> Callable[[], None]:
"""Create an idempotent callback for one already-acquired profile owner."""
-
closed = False
cleanup_lock = threading.Lock()
@@ -67,7 +64,6 @@ def _acquire_profile_runtime(
key: str,
) -> Tuple[DynamicOutputBudget, Callable[[], None]]:
"""Acquire a validated runtime and its profile owner atomically."""
-
with _runtime_lock:
runtime = _runtimes.get(key)
if runtime is not None:
@@ -85,7 +81,6 @@ def _acquire_profile_runtime(
def get_active_runtime() -> Optional[DynamicOutputBudget]:
"""Return a runtime only where an active profile loaded the plugin."""
-
key = _runtime_key()
with _runtime_lock:
if _active_profiles.get(key, 0) == 0:
@@ -106,7 +101,6 @@ def get_active_runtime() -> Optional[DynamicOutputBudget]:
def _activate_profile(key: str) -> Callable[[], None]:
"""Record one profile owner and return its idempotent release callback."""
-
with _runtime_lock:
_active_profiles[key] = _active_profiles.get(key, 0) + 1
return _profile_cleanup(key)
@@ -114,7 +108,6 @@ def _activate_profile(key: str) -> Callable[[], None]:
def _deactivate_profile(key: str) -> None:
"""Release a profile owner and invalidate its runtime after the last one."""
-
with _runtime_lock:
owners = _active_profiles.get(key, 0)
if owners <= 1:
@@ -126,7 +119,6 @@ def _deactivate_profile(key: str) -> None:
def _runtime_key() -> str:
"""Resolve Hermes' ContextVar-aware home without making it a dependency."""
-
try:
from hermes_constants import ( # type: ignore[import-not-found] # noqa: PLC0415
get_hermes_home,
@@ -145,7 +137,6 @@ def apply_incontext(
**context: Any,
) -> Optional[Dict[str, Any]]:
"""Stable function entry point used by Hermes middleware."""
-
runtime = get_active_runtime()
if runtime is None:
return None
@@ -154,7 +145,6 @@ def apply_incontext(
def register(ctx: Any) -> None:
"""Register the plugin with a Hermes ``PluginContext``."""
-
key = _runtime_key()
on_unload = getattr(ctx, "on_unload", None)
with _registration_lock:
diff --git a/incontext/preflight.py b/incontext/preflight.py
index dc0df8f..df1c891 100644
--- a/incontext/preflight.py
+++ b/incontext/preflight.py
@@ -27,7 +27,6 @@ class _ExactPressure(int):
def _live_main_route() -> Optional[Tuple[str, str, str]]:
"""Read Hermes' turn-local primary route across supported releases."""
-
try:
auxiliary = import_module("agent.auxiliary_client")
except ImportError:
@@ -53,7 +52,6 @@ def _live_main_route() -> Optional[Tuple[str, str, str]]:
def _matches_live_route(runtime: DynamicOutputBudget) -> bool:
"""Return whether the exact backend still owns Hermes' active route."""
-
route = _live_main_route()
if route is None:
return True
@@ -80,18 +78,15 @@ def __init__(
def acquire(self, owner: object, runtime: RuntimeSource) -> None:
"""Attach one installation owner and make its runtime current."""
-
self._owners.append((owner, runtime))
def release(self, owner: object) -> None:
"""Release one installation owner without disturbing the others."""
-
self._owners = [entry for entry in self._owners if entry[0] is not owner]
@property
def owned(self) -> bool:
"""Return whether this wrapper belongs to an active installation."""
-
return bool(self._owners)
def __call__(
@@ -206,18 +201,15 @@ def __init__(self, runtime: RuntimeSource, original: Callable[..., bool]) -> Non
def acquire(self, owner: object, runtime: RuntimeSource) -> None:
"""Attach one installation owner and make its runtime current."""
-
self._owners.append((owner, runtime))
def release(self, owner: object) -> None:
"""Release one installation owner without disturbing the others."""
-
self._owners = [entry for entry in self._owners if entry[0] is not owner]
@property
def owned(self) -> bool:
"""Return whether this wrapper belongs to an active installation."""
-
return bool(self._owners)
def __call__(self, *args: Any, **kwargs: Any) -> bool:
@@ -243,23 +235,19 @@ def __init__(self, original: Callable[..., bool]) -> None:
def acquire(self, owner: object) -> None:
"""Attach one installation owner."""
-
self._owners.append(owner)
def release(self, owner: object) -> None:
"""Release one installation owner without disturbing the others."""
-
self._owners = [current for current in self._owners if current is not owner]
@property
def owned(self) -> bool:
"""Return whether this wrapper belongs to an active installation."""
-
return bool(self._owners)
def __get__(self, instance: Any, owner: Any = None) -> Any:
"""Bind this callable like the Hermes instance method it replaces."""
-
del owner
return self if instance is None else MethodType(self, instance)
@@ -282,13 +270,11 @@ def __call__(
def _original_binding(binding: Any, wrapper_type: Type[Any]) -> Any:
"""Unwrap a released incontext binding before reinstalling it."""
-
return binding.original if isinstance(binding, wrapper_type) else binding
def _estimator_modules(turn_context: Any) -> Tuple[Any, ...]:
"""Return every Hermes module that owns a proactive estimator binding."""
-
try:
conversation_loop = import_module("agent.conversation_loop")
except ImportError:
@@ -307,7 +293,6 @@ def _install_estimators(
bindings: List[Any],
) -> List[InstalledBinding]:
"""Install or share exact wrappers for independent imported bindings."""
-
installed: List[InstalledBinding] = []
for module, current in zip(modules, bindings):
if isinstance(current, _ExactPreflight) and current.owned:
@@ -323,7 +308,6 @@ def _install_estimators(
def _install_exact_deferral(owner: object) -> Optional[InstalledBinding]:
"""Make Hermes' rough-only defer heuristic recognize exact pressure."""
-
try:
context_compressor = import_module("agent.context_compressor")
except ImportError:
@@ -362,7 +346,6 @@ def install(runtime: RuntimeSource) -> Optional[Cleanup]:
The original estimator remains the deliberate fail-open fallback: a
temporary tokenizer outage must not prevent Hermes from making requests.
"""
-
try:
turn_context = import_module("agent.turn_context")
except ImportError:
@@ -404,7 +387,6 @@ def install(runtime: RuntimeSource) -> Optional[Cleanup]:
def cleanup() -> None:
"""Release one owner and restore every unchanged Hermes binding."""
-
nonlocal closed
with _install_lock:
if closed:
diff --git a/incontext/settings.py b/incontext/settings.py
index fa00c65..3311faa 100644
--- a/incontext/settings.py
+++ b/incontext/settings.py
@@ -18,7 +18,6 @@ class SettingsError(RuntimeError):
def _optional_positive_integer_text(value: str) -> bool:
"""Validate an optional integer environment value after whitespace removal."""
-
if not value:
return True
try:
@@ -114,7 +113,6 @@ class Settings:
def normalize_base_url(value: Any) -> str:
"""Return a stable route identity for equivalent HTTP endpoint spellings."""
-
text = str(value or "").strip().rstrip("/")
if not text:
return ""
@@ -246,7 +244,6 @@ def _effective_compression_threshold(
compression: Mapping[str, Any],
) -> float:
"""Reuse Hermes' installed model-specific threshold policy when available."""
-
try:
from agent.auxiliary_client import ( # type: ignore[import-not-found] # noqa: PLC0415
_compression_threshold_for_model,
@@ -303,13 +300,11 @@ def _effective_compression_threshold(
def _normalized_provider_selector(value: Any) -> str:
"""Normalize the menu spelling Hermes uses for named providers."""
-
return str(value or "").strip().lower().replace(" ", "-")
def _custom_provider_aliases(display_name: Any, provider_key: Any) -> Set[str]:
"""Return normalized durable identities accepted by Hermes custom routes."""
-
aliases: Set[str] = set()
for value in (display_name, provider_key):
normalized = _normalized_provider_selector(value)
@@ -328,7 +323,6 @@ def _custom_provider_aliases(display_name: Any, provider_key: Any) -> Set[str]:
def _provider_enabled(configured: Mapping[str, Any]) -> bool:
"""Delegate optional enabled semantics to the installed Hermes release."""
-
try:
from hermes_cli.config import ( # noqa: PLC0415
is_provider_enabled,
@@ -345,13 +339,11 @@ def _provider_enabled(configured: Mapping[str, Any]) -> bool:
def _modern_provider_endpoint(configured: Mapping[str, Any]) -> Any:
"""Resolve aliases read by Hermes' direct modern-provider fast path."""
-
return configured.get("api") or configured.get("url") or configured.get("base_url")
def _valid_provider_endpoint(value: Any) -> Optional[str]:
"""Return a URL accepted by Hermes' compatibility normalizer."""
-
if not isinstance(value, str) or not value.strip():
return None
candidate = value.strip()
@@ -365,7 +357,6 @@ def _normalized_legacy_provider(
configured: Mapping[str, Any],
) -> Optional[Mapping[str, Any]]:
"""Return Hermes' canonical view of one legacy provider entry."""
-
endpoint = next(
(
candidate
@@ -391,7 +382,6 @@ def _named_provider_config(
selector: str,
) -> Optional[Mapping[str, Any]]:
"""Find a providers entry by mapping key or normalized display name."""
-
target = _normalized_provider_selector(selector)
for key, configured in providers.items():
if not isinstance(configured, Mapping):
@@ -412,7 +402,6 @@ def _compatible_modern_provider_config(
selector: str,
) -> Optional[Mapping[str, Any]]:
"""Normalize camelCase modern entries after the legacy compatibility view."""
-
target = _normalized_provider_selector(selector)
for key, configured in providers.items():
if not isinstance(configured, Mapping):
@@ -433,7 +422,6 @@ def _legacy_provider_config(
selector: str,
) -> Optional[Mapping[str, Any]]:
"""Find a saved list-style custom provider still supported by Hermes."""
-
if not isinstance(custom_providers, list):
return None
target = _normalized_provider_selector(selector)
@@ -456,7 +444,6 @@ def _configured_provider(
selector: str,
) -> Optional[Mapping[str, Any]]:
"""Resolve the new mapping before Hermes' legacy provider list."""
-
configured = _named_provider_config(providers, selector)
if configured is not None:
return configured
@@ -470,7 +457,6 @@ def _configured_provider(
def _resolved_builtin_provider(provider: str) -> Optional[str]:
"""Return Hermes' canonical built-in identity when its registry accepts it."""
-
try:
from hermes_cli.auth import ( # type: ignore[import-not-found] # noqa: PLC0415
resolve_provider,
@@ -486,7 +472,6 @@ def _resolved_builtin_provider(provider: str) -> Optional[str]:
def _effective_model_name(model: str, provider: str) -> str:
"""Mirror Hermes' provider-aware model normalization when available."""
-
try:
from hermes_cli.model_normalize import ( # type: ignore[import-not-found] # noqa: PLC0415
_AGGREGATOR_PROVIDERS,
@@ -509,7 +494,6 @@ def _effective_bare_provider(
custom_providers: Any,
) -> Tuple[str, Mapping[str, Any], bool]:
"""Resolve a non-empty, non-custom selector using Hermes' precedence."""
-
canonical = _resolved_builtin_provider(provider)
if canonical == provider:
return canonical, {}, False
@@ -529,7 +513,6 @@ def _effective_provider_route(
custom_providers: Any = None,
) -> Tuple[str, str, Mapping[str, Any]]:
"""Resolve Hermes' selector into its live provider and endpoint identity."""
-
provider_selector = _normalized_provider_selector(model.get("provider"))
provider = provider_selector
provider_config: Mapping[str, Any] = {}
@@ -589,7 +572,6 @@ def _effective_provider_route(
def _auto_uses_openai_compatible_route(base_url: Any) -> bool:
"""Mirror Hermes' explicit-local-endpoint bypass for provider auto."""
-
value = str(base_url or "").strip()
if not value:
return False
@@ -606,7 +588,6 @@ def _effective_max_tokens(
environment: HermesEnvironment,
) -> Optional[int]:
"""Resolve Hermes' output allowance in the same precedence order."""
-
if environment.max_tokens:
return int(environment.max_tokens)
configured = model.get("max_tokens")
@@ -629,7 +610,6 @@ def _validate_context_engine(
window_override: int,
) -> None:
"""Require a known boundary for non-default Hermes context engines."""
-
context_engine = str(context.get("engine") or "compressor").strip().lower()
if window_override == 0 and context_engine != "compressor":
raise SettingsError(
@@ -647,7 +627,6 @@ def _resolve_compression_window(
window_override: int,
) -> int:
"""Resolve the active automatic boundary without duplicating policy."""
-
if window_override != 0:
return window_override
if not compression_enabled:
@@ -676,7 +655,6 @@ def _validate_budget_reserves(
min_output_tokens: int,
) -> None:
"""Reject reserves that cannot leave a viable fallback request."""
-
if fallback_margin >= compression_window:
raise SettingsError(
"fallback_margin_tokens must be below the compression window",
@@ -703,7 +681,6 @@ def load_settings(
The compression window is obtained from Hermes' real ``ContextCompressor``
instead of duplicating its version-sensitive threshold arithmetic.
"""
-
if config_loader is None or compressor_class is None:
default_loader, default_compressor = _load_hermes_components()
config_loader = config_loader or default_loader
diff --git a/incontext/vllm.py b/incontext/vllm.py
index e974fa6..825aed1 100644
--- a/incontext/vllm.py
+++ b/incontext/vllm.py
@@ -92,19 +92,16 @@ def __init__(
@property
def source(self) -> str:
"""Identify successful counts without exposing endpoint details."""
-
return "vllm-tokenize"
def output_budget_field(self, requested_field: str) -> str:
"""Map Responses-only output caps to vLLM Chat Completions fields."""
-
return (
"max_tokens" if requested_field == "max_output_tokens" else requested_field
)
def coerce_output_budget(self, value: Any) -> Optional[int]:
"""Mirror vLLM's non-strict positive integer output caps."""
-
coerced = self._coerce_non_strict_integer(value)
return coerced if coerced is not None and coerced > 0 else None
@@ -115,7 +112,6 @@ def output_budget_limit(
context_length: int,
) -> Optional[int]:
"""Respect vLLM's coupled prompt-truncation/output validation."""
-
truncation_limit = self._wire_prompt_truncation_limit(request)
return None if truncation_limit is None else context_length - truncation_limit
@@ -126,7 +122,6 @@ def count(
context_length: int,
) -> int:
"""Return the exact provider-visible prompt token count."""
-
reused_prompt_tokens = self._reused_prompt_token_count(request)
self._validate_prompt_controls(request, reused_prompt_tokens)
truncation_limit = (
@@ -196,7 +191,6 @@ def count(
def clear_cache(self) -> None:
"""Discard cached counts without disturbing an in-flight request."""
-
with self._cache_lock:
self._cache_epoch += 1
self._cache.clear()
@@ -262,7 +256,6 @@ def _validate_prompt_controls(
reused_prompt_tokens: Optional[int],
) -> None:
"""Reject generation controls absent from vLLM's tokenize schema."""
-
if reused_prompt_tokens is not None:
return
extra_body = request.get("extra_body")
@@ -287,7 +280,6 @@ def _validate_prompt_controls(
@staticmethod
def _materialize_wire_value(value: Any) -> Any:
"""Copy reusable OpenAI containers into JSON-compatible shapes."""
-
if isinstance(value, Mapping):
return {
key: VllmBackend._materialize_wire_value(nested)
@@ -300,7 +292,6 @@ def _materialize_wire_value(value: Any) -> Any:
@staticmethod
def _normalize_tools(tools: Any) -> Optional[List[Any]]:
"""Materialize reusable OpenAI tool sequences for JSON tokenization."""
-
if isinstance(tools, list):
return tools
if isinstance(tools, Collection) and not isinstance(
@@ -313,7 +304,6 @@ def _normalize_tools(tools: Any) -> Optional[List[Any]]:
@staticmethod
def _normalize_messages(messages: Any) -> Any:
"""Mirror vLLM's deprecated reasoning-field normalization."""
-
if not isinstance(messages, Collection) or isinstance(
messages,
(str, bytes, Mapping),
@@ -360,7 +350,6 @@ def _nonnegative_response_integer(
key: str,
) -> int:
"""Read a response integer for a token sequence that may be empty."""
-
value = response.get(key)
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise cls.VllmBackendError(
@@ -374,7 +363,6 @@ def _prompt_truncation_limit(
request: Dict[str, Any],
) -> Optional[int]:
"""Return a safe final-prompt cap derived from vLLM truncation."""
-
truncation_limit = cls._wire_prompt_truncation_limit(request)
if truncation_limit is None or cls._has_multimodal_content(request):
# vLLM truncates rendered text token IDs before expanding media
@@ -390,7 +378,6 @@ def _wire_prompt_truncation_limit(
request: Dict[str, Any],
) -> Optional[int]:
"""Resolve vLLM's non-negative wire-level truncation constraint."""
-
extra_body = request.get("extra_body")
value = (
extra_body.get(
@@ -410,7 +397,6 @@ def _wire_prompt_truncation_limit(
@staticmethod
def _coerce_non_strict_integer(value: Any) -> Optional[int]:
"""Mirror vLLM's Pydantic integer coercion without adding a dependency."""
-
if isinstance(value, bool):
return int(value)
if isinstance(value, int):
@@ -427,7 +413,6 @@ def _coerce_non_strict_integer(value: Any) -> Optional[int]:
@classmethod
def _reused_prompt_token_count(cls, request: Dict[str, Any]) -> Optional[int]:
"""Count vLLM disaggregated-decode prompt IDs when supplied."""
-
extra_body = request.get("extra_body")
params = (
extra_body.get("kv_transfer_params", request.get("kv_transfer_params"))
@@ -458,7 +443,6 @@ def _reused_prompt_token_count(cls, request: Dict[str, Any]) -> Optional[int]:
@staticmethod
def _has_multimodal_content(request: Dict[str, Any]) -> bool:
"""Detect media-bearing messages in the provider-visible request."""
-
extra_body = request.get("extra_body")
messages = (
extra_body.get("messages", request.get("messages"))
diff --git a/incontext/vllm_provider.py b/incontext/vllm_provider.py
index 90ec2b8..6c89c4d 100644
--- a/incontext/vllm_provider.py
+++ b/incontext/vllm_provider.py
@@ -7,5 +7,4 @@
@backends.plugin("vllm", unique=True)
def provide_vllm_backend() -> Backend:
"""Construct the bundled vLLM backend on demand."""
-
return VllmBackend()
diff --git a/pyproject.toml b/pyproject.toml
index 4d87eca..bf4ec16 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "incontext"
-version = "0.0.4"
+version = "0.0.5"
authors = [{ name = "Evgeniy Blinov", email = "zheni-b@yandex.ru" }]
description = "Exact dynamic output budgeting for Hermes Agent"
readme = "README.md"
@@ -69,13 +69,26 @@ exclude_also = [
]
[tool.mypy]
-python_version = "3.8"
strict = true
packages = ["incontext"]
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+check_untyped_defs = false
+disallow_any_generics = false
+disallow_incomplete_defs = false
+disallow_subclassing_any = false
+disallow_untyped_calls = false
+disallow_untyped_decorators = false
+disallow_untyped_defs = false
+extra_checks = false
+no_implicit_reexport = false
+strict_equality = false
+warn_return_any = false
+warn_unused_ignores = false
+
[tool.ruff]
line-length = 88
-target-version = "py38"
[tool.ruff.format]
quote-style = "double"
@@ -84,13 +97,20 @@ quote-style = "double"
select = [
"A",
"ARG",
+ "ASYNC",
"B",
"BLE",
"C4",
"C90",
+ "COM818",
+ "D201",
+ "D202",
+ "D419",
"E",
+ "ERA001",
"F",
"I",
+ "INP",
"N",
"PERF",
"PIE",
@@ -101,10 +121,13 @@ select = [
"RSE",
"RUF",
"SIM",
+ "SLOT",
"T20",
+ "TID252",
"TRY",
"UP",
"W",
+ "YTT",
]
ignore = [
"TRY003",
@@ -115,7 +138,7 @@ ignore = [
]
[tool.ruff.lint.per-file-ignores]
-"tests/**/*.py" = ["ARG", "PLR0913", "PLR2004", "S101"]
+"tests/**/*.py" = ["PLR2004", "S101"]
[tool.mutmut]
paths_to_mutate = "incontext"
diff --git a/tests/e2e/test_hermes_compatibility.py b/tests/e2e/test_hermes_compatibility.py
index 67f3260..daae34e 100644
--- a/tests/e2e/test_hermes_compatibility.py
+++ b/tests/e2e/test_hermes_compatibility.py
@@ -220,7 +220,6 @@ def tokenizer_server() -> Iterator[str]:
def write_hermes_config(home: Path, inference_base_url: str) -> None:
"""Write the smallest real config that resolves all plugin invariants."""
-
home.mkdir(parents=True)
(home / "empty-bundled-plugins").mkdir()
(home / "config.yaml").write_text(
@@ -248,7 +247,6 @@ def assert_viable_output_boundary(
apply_middleware: Any,
) -> None:
"""Exercise the exact reserve boundary through real Hermes bindings."""
-
from agent import turn_context # type: ignore[import-not-found] # noqa: PLC0415
previous_prompt_tokens = TokenizerHandler.prompt_tokens
@@ -291,7 +289,6 @@ def assert_in_turn_tool_growth_uses_exact_pressure(
make that binding exact and prevent a recent-compaction marker from
suppressing the authoritative result before request middleware runs.
"""
-
from agent import ( # noqa: PLC0415
conversation_loop, # type: ignore[import-not-found]
)
@@ -322,7 +319,6 @@ def assert_in_turn_tool_growth_uses_exact_pressure(
def make_history(*, pairs: int = 30, width: int = 1200) -> list[dict[str, Any]]:
"""Build enough alternating turns for the real compressor to operate."""
-
history: list[dict[str, Any]] = []
for index in range(pairs):
history.extend(
@@ -344,7 +340,6 @@ def build_test_agent(
inference_base_url: str,
) -> tuple[Any, list[tuple[str, str]]]:
"""Construct a side-effect-free real Hermes agent for one scenario."""
-
status_messages: list[tuple[str, str]] = []
with patch(
"run_agent.get_tool_definitions",
@@ -387,7 +382,6 @@ def run_test_agent(
history: list[dict[str, Any]],
) -> dict[str, Any]:
"""Run the real turn loop while suppressing unrelated persistence effects."""
-
with patch.object(
agent,
"_persist_session",
@@ -410,7 +404,6 @@ def run_test_agent(
def assert_no_tiny_output_cap(request: dict[str, Any], runtime: Any) -> None:
"""Allow fail-open provider defaults, but reject zero or tiny sentinels."""
-
for field in ("max_tokens", "max_completion_tokens", "max_output_tokens"):
value = request.get(field)
if value is not None:
@@ -424,7 +417,6 @@ def assert_complete_auto_compression(
inference_base_url: str,
) -> None:
"""Run an oversized turn through real Hermes compression and inference."""
-
token_request_start = len(TokenizerHandler.requests)
chat_request_start = len(TokenizerHandler.chat_requests)
oversized_prompt_tokens = (
@@ -491,7 +483,6 @@ def assert_repeated_auto_compression(
inference_base_url: str,
) -> None:
"""Require a second real summary before allowing the main inference."""
-
TokenizerHandler.chat_requests.clear()
token_request_start = len(TokenizerHandler.requests)
chat_request_start = 0
@@ -558,7 +549,6 @@ def assert_no_progress_stops_compression(
Both behaviours are safe only when they remain bounded and do not replace
the viable-output reserve with a tiny sentinel.
"""
-
TokenizerHandler.chat_requests.clear()
chat_request_start = 0
compression_calls = 0
@@ -576,7 +566,7 @@ def no_progress(
compression_calls += 1
return messages, "You are the incontext e2e agent."
- TokenizerHandler.count_resolver = lambda payload: oversized_prompt_tokens
+ TokenizerHandler.count_resolver = lambda _payload: oversized_prompt_tokens
runtime.backend.clear_cache()
history = make_history()
try:
@@ -612,7 +602,6 @@ def assert_compression_attempt_limit_is_bounded(
Exact pressure may activate both older guards, but must never bypass their
combined upper bound or emit a zero-token completion sentinel afterwards.
"""
-
TokenizerHandler.chat_requests.clear()
chat_request_start = 0
compression_calls = 0
@@ -669,7 +658,6 @@ def assert_summary_error_is_bounded(
inference_base_url: str,
) -> None:
"""Preserve the turn when the real summary request is rejected."""
-
TokenizerHandler.chat_requests.clear()
chat_request_start = 0
oversized_prompt_tokens = (
@@ -719,12 +707,11 @@ def assert_summary_timeout_is_bounded(
inference_base_url: str,
) -> None:
"""Bound a real compressor timeout without emitting an unusable cap."""
-
TokenizerHandler.chat_requests.clear()
oversized_prompt_tokens = (
runtime.settings.compression_window - runtime.settings.min_output_tokens + 1
)
- TokenizerHandler.count_resolver = lambda payload: oversized_prompt_tokens
+ TokenizerHandler.count_resolver = lambda _payload: oversized_prompt_tokens
runtime.backend.clear_cache()
try:
agent, _ = build_test_agent(inference_base_url)
@@ -753,7 +740,6 @@ def assert_summary_timeout_is_bounded(
def assert_compression_scenarios(runtime: Any, inference_base_url: str) -> None:
"""Exercise successful, repeated, and bounded failure outcomes."""
-
assert_complete_auto_compression(runtime, inference_base_url)
assert_repeated_auto_compression(runtime, inference_base_url)
assert_no_progress_stops_compression(runtime, inference_base_url)
@@ -768,7 +754,6 @@ def test_pypi_entrypoint_runs_the_complete_hermes_compression_path(
tokenizer_server: str,
) -> None:
"""Load through metadata, register, and execute through Hermes itself."""
-
home = tmp_path / "hermes"
inference_base_url = tokenizer_server.rsplit("/", 1)[0] + "/v1"
write_hermes_config(home, inference_base_url)
diff --git a/tests/units/conftest.py b/tests/units/conftest.py
index 5014d12..fe518f8 100644
--- a/tests/units/conftest.py
+++ b/tests/units/conftest.py
@@ -9,7 +9,6 @@
@pytest.fixture(autouse=True)
def reset_skelet_environment_caches() -> None:
"""Keep skelet's process-environment cache isolated between unit tests."""
-
for storage in (Environment, HermesEnvironment, VllmEnvironment):
for source in storage.__sources__.sources:
source.__dict__.pop("data", None)
diff --git a/tests/units/test_auxiliary.py b/tests/units/test_auxiliary.py
index b385ea6..0ce2e30 100644
--- a/tests/units/test_auxiliary.py
+++ b/tests/units/test_auxiliary.py
@@ -56,7 +56,7 @@ def install_fake_hermes(
agent.__path__ = [] # type: ignore[attr-defined]
auxiliary = types.ModuleType("agent.auxiliary_client")
- def build(
+ def build( # noqa: PLR0913
provider: str,
model: str,
messages: list[Any],
@@ -150,7 +150,6 @@ def test_auxiliary_uses_hermes_provider_output_alias(
field, so both bounded and unbounded calls must seed dynamic budgeting with
``max_completion_tokens`` when Hermes selects it.
"""
-
auxiliary, _ = install_fake_hermes(monkeypatch)
selections: list[tuple[int, str]] = []
@@ -177,8 +176,8 @@ def select(value: int, *, model: str) -> dict[str, int]:
@pytest.mark.parametrize(
"selector",
[
- lambda value, **context: {"max_tokens": False},
- lambda value, **context: (_ for _ in ()).throw(RuntimeError("boom")),
+ lambda _value, **_context: {"max_tokens": False},
+ lambda _value, **_context: (_ for _ in ()).throw(RuntimeError("boom")),
],
)
def test_auxiliary_falls_back_from_an_invalid_hermes_output_selector(
@@ -220,7 +219,6 @@ def test_auxiliary_wrapper_accepts_additive_hermes_parameters(
once Hermes has built the request, that metadata drift must only skip
budgeting rather than fail the otherwise valid auxiliary call.
"""
-
auxiliary, builder = install_fake_hermes(monkeypatch)
if stale_signature:
builder.__signature__ = Signature( # type: ignore[attr-defined]
@@ -313,7 +311,6 @@ def test_auxiliary_wrapper_ignores_a_different_fallback_model() -> None:
Applying that tokenizer and context window to the fallback would corrupt
its request; the original provider kwargs must pass through untouched.
"""
-
counter = Counter(12_345)
def build(
@@ -347,7 +344,6 @@ def test_auxiliary_wrapper_ignores_same_model_on_another_route() -> None:
still wrong when the request is headed to an external provider that happens
to expose the same alias.
"""
-
counter = Counter(12_345)
scoped_runtime = DynamicOutputBudget(
Settings(
@@ -394,7 +390,6 @@ def test_auxiliary_budgets_synthetic_main_agent_fallback_label() -> None:
deployment exactly; rejecting only the synthetic label loses both exact
tokenization and the bounded summary cap that Hermes omitted upstream.
"""
-
counter = Counter(12_345)
scoped_runtime = DynamicOutputBudget(
Settings(
@@ -438,7 +433,6 @@ def test_auxiliary_wrapper_ignores_same_endpoint_on_another_provider() -> None:
contracts and model routing. Matching only the URL and model would still
let the primary backend rewrite a fallback request owned by another route.
"""
-
counter = Counter(12_345)
scoped_runtime = DynamicOutputBudget(
Settings(
@@ -481,7 +475,6 @@ def test_auxiliary_accepts_canonical_equivalent_route() -> None:
does not change route identity, so those transformations must not silently
bypass exact budgeting for the configured primary endpoint.
"""
-
counter = Counter(12_345)
scoped_runtime = DynamicOutputBudget(
Settings(
@@ -523,7 +516,6 @@ def test_auxiliary_runtime_resolver_tracks_the_active_profile() -> None:
in 2026.8. The resolver must therefore be invoked for every request so a
profile switch cannot retain the previous profile's backend and window.
"""
-
active = runtime(Counter(12_345))
calls: list[None] = []
@@ -679,7 +671,6 @@ def test_latest_auxiliary_owner_cleanup_restores_previous_runtime(
removed runtime would tokenize later requests with a stale model, backend,
or compression window even though its profile no longer owns the plugin.
"""
-
auxiliary, _ = install_fake_hermes(monkeypatch)
first = Counter(10_000)
second = Counter(20_000)
@@ -716,7 +707,6 @@ def test_cleanup_restores_builder_after_the_last_plugin_owner(
other, while the final callback must conditionally restore the exact
original builder and remain safe if invoked twice.
"""
-
auxiliary, original = install_fake_hermes(monkeypatch)
first_cleanup = install(runtime(Counter(100)))
second_cleanup = install(runtime(Counter(200)))
@@ -742,14 +732,15 @@ def test_cleanup_never_overwrites_a_later_auxiliary_wrapper(
callable must survive incontext unload; cleanup only releases internal
ownership and must never put an older function back over newer state.
"""
-
auxiliary, original = install_fake_hermes(monkeypatch)
first = Counter(100)
cleanup = install(runtime(first))
assert callable(cleanup)
stale_wrapper = auxiliary._build_call_kwargs # type: ignore[attr-defined]
- replacement = lambda *args, **kwargs: {} # noqa: E731
+ def replacement(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+ return {}
+
auxiliary._build_call_kwargs = replacement # type: ignore[attr-defined]
cleanup()
@@ -781,11 +772,12 @@ def test_final_owner_release_cannot_race_auxiliary_configuration() -> None:
tuples and either raise or mix profile configuration; the in-flight call
must retain the complete pre-cleanup snapshot.
"""
-
checked = threading.Event()
resume = threading.Event()
class BlockingOwners(tuple):
+ __slots__ = ()
+
def __bool__(self) -> bool:
checked.set()
assert resume.wait(timeout=2)
@@ -833,7 +825,6 @@ def test_stale_auxiliary_cleanup_cannot_remove_a_new_installation(
decrement or restore the new installation's reference count, but it must
still release its own owner and restore its now-detached module.
"""
-
stale_auxiliary, stale_original = install_fake_hermes(monkeypatch)
stale_cleanup = install(runtime(Counter(100)))
assert callable(stale_cleanup)
@@ -875,7 +866,6 @@ def test_install_skips_a_missing_private_builder(
layer, so a missing private binding must warn and return ``None`` rather
than raising ``KeyError`` and rolling back the stable public middleware.
"""
-
agent = types.ModuleType("agent")
agent.__path__ = [] # type: ignore[attr-defined]
auxiliary = types.ModuleType("agent.auxiliary_client")
diff --git a/tests/units/test_backend.py b/tests/units/test_backend.py
index e38b7ee..f790865 100644
--- a/tests/units/test_backend.py
+++ b/tests/units/test_backend.py
@@ -1,6 +1,6 @@
from __future__ import annotations
-from typing import Any, get_type_hints
+from typing import Any, cast, get_type_hints
from unittest.mock import patch
import pytest
@@ -25,6 +25,7 @@ def count(
*,
context_length: int,
) -> int:
+ del request
return context_length
def clear_cache(self) -> None:
@@ -33,7 +34,6 @@ def clear_cache(self) -> None:
def test_backend_contract_is_public_and_abstract() -> None:
"""Expose the extension contract at the documented package boundary."""
-
assert incontext.Backend is Backend
assert incontext.backends is backends
with pytest.raises(TypeError):
@@ -47,7 +47,6 @@ def test_backend_preserves_output_budget_alias_by_default() -> None:
fields remain untouched unless a backend explicitly documents a wire-level
incompatibility such as vLLM Chat Completions' ignored Responses alias.
"""
-
assert ReplacementBackend().output_budget_field("max_output_tokens") == (
"max_output_tokens"
)
@@ -61,7 +60,6 @@ def test_backend_adds_no_provider_output_limit_by_default() -> None:
default extension must therefore leave their dynamic budget untouched
until a backend explicitly reports an additional limit.
"""
-
assert (
ReplacementBackend().output_budget_limit(
{"model": "replacement", "messages": []},
@@ -80,9 +78,8 @@ def test_public_type_hints_resolve_on_every_supported_python() -> None:
callable must resolve at runtime on each interpreter declared in package
metadata, not merely parse successfully there.
"""
-
targets = (
- Backend.source.fget,
+ cast(property, Backend.__dict__["source"]).fget,
Backend.count,
Backend.clear_cache,
Backend.output_budget_field,
@@ -94,7 +91,7 @@ def test_public_type_hints_resolve_on_every_supported_python() -> None:
apply_incontext,
register,
VllmBackend.__init__,
- VllmBackend.source.fget,
+ cast(property, VllmBackend.__dict__["source"]).fget,
VllmBackend.count,
VllmBackend.clear_cache,
VllmBackend.output_budget_field,
@@ -129,7 +126,6 @@ def test_named_backend_can_replace_the_bundled_backend() -> None:
uniqueness policy. A second distribution using the same selected name
must fail during discovery instead of making startup order-dependent.
"""
-
expected = ReplacementBackend()
@backends.plugin("unit_replacement")
diff --git a/tests/units/test_budget.py b/tests/units/test_budget.py
index efa1a14..c16bbf0 100644
--- a/tests/units/test_budget.py
+++ b/tests/units/test_budget.py
@@ -191,7 +191,6 @@ def test_rough_fallback_counts_extra_body_prompt_overrides(
outage; budgeting from superseded top-level content can otherwise allocate
output beyond the compression boundary by an unbounded amount.
"""
-
top_messages = [{"role": "user", "content": "superseded"}]
wire_messages = [{"role": "user", "content": f"wire-{index}"} for index in range(7)]
wire_tools = [{"type": "function", "function": {"name": "wire"}}]
@@ -239,7 +238,7 @@ def test_runtime_exact_count_preserves_smallest_existing_output_cap(
runtime = budget.DynamicOutputBudget(
runtime_settings,
counter,
- rough_estimator=lambda request: 999,
+ rough_estimator=lambda _request: 999,
)
request = {
"model": "qwen",
@@ -349,7 +348,6 @@ def test_runtime_budgets_reusable_message_collections(
reusable sequence skips the dynamic cap for an otherwise valid request;
incontext must copy it without mutating the caller-owned collection.
"""
-
messages = ({"role": "user", "content": "tuple prompt"},)
counter = Counter(10_000)
runtime = budget.DynamicOutputBudget(runtime_settings, counter)
@@ -373,7 +371,6 @@ def test_runtime_cleans_output_caps_from_read_only_extra_body_mapping(
wire. The rewrite must copy and clean the mapping without mutating the
caller-owned object.
"""
-
nested_messages = ({"role": "user", "content": "wire prompt"},)
nested = MappingProxyType(
{
@@ -416,7 +413,6 @@ def test_runtime_rejects_model_override_in_read_only_extra_body_mapping(
with the primary backend and context window even though a different model
is provider-visible.
"""
-
counter = Counter(100)
runtime = budget.DynamicOutputBudget(runtime_settings, counter)
@@ -443,7 +439,6 @@ def test_runtime_accepts_an_exact_zero_token_truncated_prompt(
that case; treating zero as an invalid estimate crashes middleware before
it can preserve the provider's full safe output allowance.
"""
-
runtime = budget.DynamicOutputBudget(runtime_settings, Counter(0))
result = runtime(request={"model": "qwen", "messages": []})
@@ -576,7 +571,6 @@ def test_runtime_preserves_the_provider_selected_output_field(field: str) -> Non
use their own alias; replacing the chosen name can turn a valid request into
HTTP 400 even when the numeric dynamic budget is correct.
"""
-
runtime_settings = Settings(
model_name="qwen",
context_length=65_536,
@@ -697,7 +691,6 @@ def test_runtime_removes_extra_body_output_cap_override() -> None:
when selecting the smallest caller cap, move the safe result to the same
top-level field, and remove every nested alias without mutating the input.
"""
-
runtime_settings = Settings(
model_name="qwen",
context_length=1000,
@@ -741,7 +734,6 @@ def test_nested_smaller_cap_preserves_top_level_provider_field() -> None:
but removing that override must emit the minimum through the existing
top-level provider field rather than reintroducing the rejected alias.
"""
-
runtime_settings = Settings(
model_name="gpt-5-test",
context_length=1000,
@@ -779,7 +771,6 @@ def test_runtime_rejects_extra_body_model_override(
bundled backend; otherwise a fallback model receives the primary model's
token count and compression-window arithmetic.
"""
-
counter = Counter(100)
runtime = budget.DynamicOutputBudget(runtime_settings, counter)
request = {
@@ -903,7 +894,7 @@ def test_runtime_fallback_reserves_safety_margin(
runtime = budget.DynamicOutputBudget(
runtime_settings,
Counter(TimeoutError("secret failure")),
- rough_estimator=lambda request: 12_000,
+ rough_estimator=lambda _request: 12_000,
)
request = {
"model": "qwen",
@@ -942,7 +933,7 @@ def test_runtime_normalizes_non_positive_fallback_estimate(
runtime = budget.DynamicOutputBudget(
runtime_settings,
Counter(RuntimeError()),
- rough_estimator=lambda request: 0,
+ rough_estimator=lambda _request: 0,
)
result = runtime(request={"model": "qwen", "messages": []})
assert result is not None
@@ -953,7 +944,7 @@ def test_runtime_leaves_request_unchanged_when_both_estimators_fail(
runtime_settings: Settings,
caplog: pytest.LogCaptureFixture,
) -> None:
- def broken_fallback(request: dict[str, Any]) -> int:
+ def broken_fallback(_request: dict[str, Any]) -> int:
raise ValueError("fallback secret")
runtime = budget.DynamicOutputBudget(
@@ -990,7 +981,6 @@ def test_runtime_ignores_request_for_a_different_model(
compression boundary for that model, fail-open is safer than computing an
apparently exact cap from the primary model's tokenizer.
"""
-
counter = Counter(100)
runtime = budget.DynamicOutputBudget(runtime_settings, counter)
@@ -1013,7 +1003,6 @@ def test_runtime_ignores_same_model_on_a_different_provider_route() -> None:
incontext must not apply the primary vLLM tokenizer merely because the JSON
model string still matches.
"""
-
counter = Counter(100)
runtime = budget.DynamicOutputBudget(
Settings(
@@ -1048,7 +1037,6 @@ def test_runtime_accepts_canonical_equivalent_route() -> None:
port, and a trailing slash must compare as one route so exact budgeting is
not accidentally disabled for the configured endpoint.
"""
-
counter = Counter(100)
runtime = budget.DynamicOutputBudget(
Settings(
diff --git a/tests/units/test_hermes.py b/tests/units/test_hermes.py
index 8695ecd..9978ebb 100644
--- a/tests/units/test_hermes.py
+++ b/tests/units/test_hermes.py
@@ -4,6 +4,7 @@
import sys
import threading
import types
+from collections.abc import Iterator
from pathlib import Path
from typing import Any
from unittest import mock
@@ -29,7 +30,7 @@ def on_unload(self, callback: Any) -> None:
@pytest.fixture(autouse=True)
-def reset_runtime() -> None:
+def reset_runtime() -> Iterator[None]:
hermes._reset_runtime_for_tests()
yield
hermes._reset_runtime_for_tests()
@@ -83,7 +84,6 @@ def test_get_runtime_isolated_by_active_hermes_home() -> None:
Returning to profile A must reuse A's backend, while profile B receives a
separately constructed settings/backend pair instead of inheriting A.
"""
-
first = mock.create_autospec(DynamicOutputBudget, instance=True)
second = mock.create_autospec(DynamicOutputBudget, instance=True)
keys = iter(["profile-a", "profile-b", "profile-a"])
@@ -107,7 +107,6 @@ def test_active_runtime_is_scoped_to_registered_profiles() -> None:
B instead of lazily creating B's runtime and sending its prompt to A's
configured tokenizer endpoint.
"""
-
runtime = mock.create_autospec(DynamicOutputBudget, instance=True)
with mock.patch.object(
hermes,
@@ -126,7 +125,6 @@ def test_active_runtime_builds_after_profile_activation() -> None:
the profile remains registered. The next wrapped call must rebuild only
that active profile instead of treating it as disabled.
"""
-
runtime = mock.create_autospec(DynamicOutputBudget, instance=True)
with mock.patch.object(
hermes,
@@ -145,7 +143,6 @@ def test_profile_runtime_survives_until_its_last_owner_unloads() -> None:
unload from either manager must decrement ownership without invalidating a
runtime that the remaining manager and its in-flight requests still use.
"""
-
runtime = mock.create_autospec(DynamicOutputBudget, instance=True)
hermes._runtimes["profile-a"] = runtime
hermes._activate_profile("profile-a")
@@ -169,7 +166,6 @@ def test_concurrent_registration_retains_its_validated_runtime(
rebuild and may fail. Runtime acquisition and owner increment must share
one critical section, leaving no observable owner-without-runtime state.
"""
-
runtime = mock.Mock()
runtime.settings = runtime_settings
first_context = Context()
@@ -244,7 +240,6 @@ def test_profile_runtime_acquisition_closes_the_final_unload_gap(
section and proves the new owner is visible before unload can invalidate
the runtime.
"""
-
runtime = mock.Mock()
runtime.settings = runtime_settings
real_lock = threading.Lock()
@@ -304,7 +299,6 @@ def test_profile_runtime_builder_can_query_active_runtime_without_deadlock(
code blocks forever. The finished runtime and owner must still become
visible together after construction.
"""
-
runtime = mock.Mock()
runtime.settings = runtime_settings
context = Context()
@@ -349,7 +343,6 @@ def test_concurrent_runtime_publication_keeps_the_existing_winner() -> None:
lock and retain the first published object rather than overwrite the
runtime already used by another request or profile owner.
"""
-
candidate = mock.create_autospec(DynamicOutputBudget, instance=True)
winner = mock.create_autospec(DynamicOutputBudget, instance=True)
@@ -381,7 +374,6 @@ def test_active_runtime_discards_candidate_if_owner_unloads_during_build() -> No
recheck ownership and discard the candidate when cleanup wins; otherwise a
removed profile's old settings are cached for the next forced discovery.
"""
-
candidate = mock.create_autospec(DynamicOutputBudget, instance=True)
hermes._active_profiles["profile-a"] = 1
@@ -405,7 +397,6 @@ def _capture_registration_failure(
failures: list[BaseException],
) -> None:
"""Record an unexpected thread exception for deterministic assertions."""
-
try:
callback(context)
except BaseException as exc: # noqa: BLE001
@@ -421,7 +412,6 @@ def test_profile_cleanup_is_concurrently_idempotent() -> None:
the runtime owned by another manager. A per-callback lock must preserve
that independent owner under a synchronized burst of duplicate calls.
"""
-
runtime = mock.create_autospec(DynamicOutputBudget, instance=True)
hermes._runtimes["profile-a"] = runtime
cleanup = hermes._activate_profile("profile-a")
@@ -450,7 +440,6 @@ def test_profile_unload_invalidates_its_cached_runtime() -> None:
Once all owners have unloaded, retaining the previous cached runtime would
silently preserve the old context window and tokenizer route on reload.
"""
-
first = mock.create_autospec(DynamicOutputBudget, instance=True)
second = mock.create_autospec(DynamicOutputBudget, instance=True)
first.settings = mock.Mock(context_length=65_536, compression_window=64_000)
@@ -497,7 +486,6 @@ def test_runtime_key_uses_context_aware_hermes_home(
enters another home. Importing ``get_hermes_home`` lazily and resolving its
path mirrors Hermes' own profile-scoped PluginManager cache.
"""
-
constants = types.ModuleType("hermes_constants")
constants.get_hermes_home = lambda: tmp_path / "nested" / ".." # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "hermes_constants", constants)
@@ -514,7 +502,6 @@ def test_runtime_key_falls_back_without_hermes_installation(
Hermes environment. Failure to import ``hermes_constants`` must map to the
single default runtime key instead of making ``get_runtime`` unimportable.
"""
-
original_import = builtins.__import__
def rejecting_import(name: str, *args: Any, **kwargs: Any) -> Any:
@@ -574,7 +561,6 @@ def test_stale_middleware_during_unload_does_not_resurrect_runtime(
runtime; otherwise force rediscovery reuses settings constructed during
unload rather than loading the profile's new configuration.
"""
-
first = mock.Mock()
first.settings = runtime_settings
fresh = mock.Mock()
@@ -746,7 +732,6 @@ def test_failed_legacy_force_reload_releases_immortal_installation(
forever. All original acquisitions must therefore be released in reverse
order and the cached runtime invalidated.
"""
-
events: list[str] = []
class LegacyContext:
@@ -798,7 +783,6 @@ def test_register_skips_missing_cleanup_callbacks(
Their installers return ``None``; the public middleware should still load
without handing invalid callbacks to the ownership ledger.
"""
-
runtime = mock.Mock()
runtime.settings = runtime_settings
context = Context()
@@ -831,7 +815,6 @@ def test_register_rolls_back_when_auxiliary_installation_fails(
Hermes never receives unload callbacks for a plugin whose registration
failed.
"""
-
runtime = mock.Mock()
runtime.settings = runtime_settings
preflight_cleanup = mock.Mock()
@@ -872,7 +855,6 @@ def test_register_rolls_back_all_integrations_when_middleware_rejects(
acquisition order and remain independent of the absent unload ledger so no
process-wide patch continues handling requests for a disabled profile.
"""
-
events: list[str] = []
class RejectingContext(Context):
diff --git a/tests/units/test_preflight.py b/tests/units/test_preflight.py
index f4dc862..5868512 100644
--- a/tests/units/test_preflight.py
+++ b/tests/units/test_preflight.py
@@ -66,7 +66,7 @@ def install_fake_hermes(
loop.estimate_request_tokens_rough = original # type: ignore[attr-defined]
turn_context.estimate_request_tokens_rough = original # type: ignore[attr-defined]
turn_context._should_run_preflight_estimate = ( # type: ignore[attr-defined]
- lambda messages, protect_first_n, protect_last_n, threshold_tokens: False
+ lambda messages, protect_first_n, protect_last_n, threshold_tokens: False # noqa: ARG005
)
monkeypatch.setitem(sys.modules, "agent", agent)
monkeypatch.setitem(sys.modules, "agent.conversation_loop", loop)
@@ -94,6 +94,7 @@ def rough(
system_prompt: str = "",
tools: Any = None,
) -> int:
+ del messages, system_prompt, tools
raise AssertionError("exact backend should be used")
loop, turn_context = install_fake_hermes(monkeypatch, rough)
@@ -140,7 +141,7 @@ def should_defer_preflight_to_real_usage(self, tokens: int) -> bool:
monkeypatch.setitem(sys.modules, "agent.context_compressor", compressor_module)
loop, _ = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 30_508,
+ lambda messages, *, system_prompt="", tools=None: 30_508, # noqa: ARG005
)
first_cleanup = install(runtime(Counter(61_052)))
second_cleanup = install(runtime(Counter(61_052)))
@@ -176,10 +177,9 @@ def test_install_keeps_turn_preflight_when_in_turn_hook_is_unavailable(
may temporarily expose no callable under the private binding. Neither API
shape should disable exact compression at the still-supported turn start.
"""
-
_, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 7,
+ lambda messages, *, system_prompt="", tools=None: 7, # noqa: ARG005
)
if loop_api is None:
monkeypatch.delitem(sys.modules, "agent.conversation_loop")
@@ -202,13 +202,12 @@ def test_install_tolerates_hermes_without_the_rough_defer_hook(
bindings active instead of turning a compatibility optimization into a
mandatory private API dependency.
"""
-
compressor_module = types.ModuleType("agent.context_compressor")
compressor_module.ContextCompressor = None # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "agent.context_compressor", compressor_module)
loop, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 7,
+ lambda messages, *, system_prompt="", tools=None: 7, # noqa: ARG005
)
cleanup = install(runtime(Counter(123)))
@@ -258,8 +257,7 @@ def test_preflight_counts_provider_visible_api_content_without_mutation(
undercount injected memory or plugin context and skip compression. The
preflight copy must mirror substitution while preserving retry-owned input.
"""
-
- messages = [
+ messages: list[Any] = [
{
"role": "user",
"content": "clean",
@@ -274,7 +272,7 @@ def test_preflight_counts_provider_visible_api_content_without_mutation(
]
_, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 1,
+ lambda messages, *, system_prompt="", tools=None: 1, # noqa: ARG005
)
counter = Counter(321)
install(runtime(counter))
@@ -379,7 +377,7 @@ def test_preflight_pressure_matches_the_viable_output_boundary(
) -> None:
_, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 1,
+ lambda messages, *, system_prompt="", tools=None: 1, # noqa: ARG005
)
install(runtime(Counter(prompt_tokens)))
@@ -402,7 +400,6 @@ def test_configured_output_reserve_drives_preflight_and_middleware_together(
the default 4096-token reserve. Exercise the smallest legal reserve, a
custom ordinary value, and the largest reserve below this test window.
"""
-
window = 100
prompt_tokens = window - minimum_output_tokens + shortfall
configured = Settings(
@@ -418,7 +415,7 @@ def test_configured_output_reserve_drives_preflight_and_middleware_together(
active = DynamicOutputBudget(configured, counter)
_, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 1,
+ lambda messages, *, system_prompt="", tools=None: 1, # noqa: ARG005
)
install(active)
@@ -445,7 +442,6 @@ def test_fallback_preflight_and_middleware_share_the_exact_boundary(
shortfall: int,
) -> None:
"""Prove ``W - P - F == R`` is viable and one token less compresses."""
-
window = 100
rough_prompt_tokens = (
window - fallback_margin_tokens - minimum_output_tokens + shortfall
@@ -472,7 +468,7 @@ def rough(
active = DynamicOutputBudget(
configured,
Counter(TimeoutError("exact tokenizer unavailable")),
- rough_estimator=lambda request: rough_prompt_tokens,
+ rough_estimator=lambda _request: rough_prompt_tokens,
)
_, turn_context = install_fake_hermes(monkeypatch, rough)
install(active)
@@ -498,7 +494,6 @@ def test_live_route_reader_supports_legacy_hermes_globals(
live-switch protection active across the package's declared compatibility
range instead of silently trusting a stale tokenizer on the older release.
"""
-
auxiliary = types.ModuleType("agent.auxiliary_client")
auxiliary._RUNTIME_MAIN_PROVIDER = "custom" # type: ignore[attr-defined]
auxiliary._RUNTIME_MAIN_MODEL = "qwen-test" # type: ignore[attr-defined]
@@ -516,7 +511,6 @@ def test_live_route_reader_marks_a_broken_private_api_as_unmatched(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Fail to Hermes' rough estimator when private route lookup breaks."""
-
auxiliary = types.ModuleType("agent.auxiliary_client")
def fail(field: str) -> str:
@@ -550,7 +544,6 @@ def test_preflight_route_guard_checks_provider_and_endpoint(
exact preflight must require all configured route dimensions to match. DNS
case, a default HTTPS port, and a trailing slash remain the same endpoint.
"""
-
auxiliary = types.ModuleType("agent.auxiliary_client")
live_route = {
"provider": provider,
@@ -577,7 +570,6 @@ def test_preflight_route_guard_checks_provider_and_endpoint(
def test_exact_gate_resolves_profile_runtime_at_call_time() -> None:
"""Use the active profile's resolver for every cheap-gate decision."""
-
active = runtime(Counter(1))
calls: list[None] = []
@@ -585,7 +577,7 @@ def resolve() -> DynamicOutputBudget:
calls.append(None)
return active
- gate = _ExactPreflightGate(resolve, lambda *args, **kwargs: False)
+ gate = _ExactPreflightGate(resolve, lambda *_args, **_kwargs: False)
assert gate([], 3, 20, 64_000) is True
assert calls == [None]
@@ -596,7 +588,7 @@ def test_preflight_preserves_empty_tool_shape(
) -> None:
_, turn_context = install_fake_hermes(
monkeypatch,
- lambda messages, *, system_prompt="", tools=None: 42,
+ lambda messages, *, system_prompt="", tools=None: 42, # noqa: ARG005
)
counter = Counter(123)
install(runtime(counter))
@@ -677,7 +669,6 @@ def test_preflight_fails_open_for_additive_hermes_estimator_context(
the complete call to the original estimator instead of raising before the
request path can proceed.
"""
-
calls: list[tuple[Any, str, Any, Any]] = []
def rough(
@@ -717,6 +708,7 @@ def rough(
system_prompt: str = "",
tools: Any = None,
) -> int:
+ del messages, system_prompt, tools
return 5
_, turn_context = install_fake_hermes(monkeypatch, rough)
@@ -857,7 +849,6 @@ def test_preflight_runtime_resolver_tracks_the_active_profile() -> None:
home is ContextVar-scoped. Resolving lazily ensures the wrapper uses
the runtime belonging to the profile that initiated this particular turn.
"""
-
active = runtime(Counter(456))
calls: list[None] = []
@@ -888,7 +879,6 @@ def test_preflight_skips_profiles_without_an_active_plugin_owner() -> None:
both exact tokenization and any cross-profile prompt disclosure while a
different profile keeps the shared wrapper installed.
"""
-
counter = Counter(999)
def rough(
@@ -925,7 +915,6 @@ def test_final_owner_release_cannot_race_active_preflight(is_gate: bool) -> None
list between truthiness and indexing on free-threaded CPython. One snapshot
must instead remain valid for the complete lookup in both wrappers.
"""
-
checked = threading.Event()
resume = threading.Event()
@@ -946,7 +935,7 @@ def rough(
resolver = lambda: None # noqa: E731
wrapper = (
- _ExactPreflightGate(resolver, lambda *args, **kwargs: False)
+ _ExactPreflightGate(resolver, lambda *_args, **_kwargs: False)
if is_gate
else _ExactPreflight(resolver, rough)
)
diff --git a/tests/units/test_settings.py b/tests/units/test_settings.py
index 1a63678..5b6caf7 100644
--- a/tests/units/test_settings.py
+++ b/tests/units/test_settings.py
@@ -5,7 +5,7 @@
import types
from collections.abc import Mapping
from dataclasses import FrozenInstanceError
-from typing import Any, ClassVar
+from typing import Any, ClassVar, cast
from unittest.mock import patch
import pytest
@@ -13,7 +13,7 @@
from incontext import settings
base_environment: dict[str, str] = {}
-base_config = {
+base_config: dict[str, Any] = {
"model": {
"default": "qwen-test",
"context_length": 65_536,
@@ -28,7 +28,7 @@
class ModernCompressor:
calls: ClassVar[list[dict[str, Any]]] = []
- def __init__(
+ def __init__( # noqa: PLR0913
self,
model: str,
threshold_percent: float,
@@ -67,12 +67,11 @@ def test_settings_remains_slotted_on_python_38() -> None:
validation. Retaining both slots and the frozen dataclass guard prevents a
later field assignment from silently changing only part of that snapshot.
"""
-
loaded = load()
assert not hasattr(loaded, "__dict__")
with pytest.raises(FrozenInstanceError):
- loaded.model_name = "changed-after-validation"
+ cast(Any, loaded).model_name = "changed-after-validation"
@pytest.mark.parametrize(("value", "expected"), [(1, 1), (" 42 ", 42), (0, 0)])
@@ -128,7 +127,6 @@ def test_strict_float_wraps_unrepresentable_integer() -> None:
configured threshold must fail as ``SettingsError`` like every other
malformed numeric value instead of leaking ``OverflowError`` from startup.
"""
-
with pytest.raises(settings.SettingsError, match="must be numeric"):
settings._strict_float(10**10_000, "value", minimum_exclusive=0)
@@ -203,7 +201,6 @@ def test_load_settings_uses_configured_minimum_output_reserve() -> None:
def test_load_settings_accepts_the_last_viable_fallback_policy() -> None:
"""Allow the equality edge that still leaves one token for the prompt."""
-
result = load(
environment={
"INCONTEXT_COMPRESSION_WINDOW_TOKENS": "5000",
@@ -224,7 +221,6 @@ def test_load_settings_rejects_a_fallback_policy_with_no_viable_prompt() -> None
so fallback preflight would request compression forever without a viable
post-compression budget.
"""
-
with pytest.raises(
settings.SettingsError,
match="fallback_margin_tokens plus min_output_tokens must be below",
@@ -258,7 +254,6 @@ def test_normalized_model_thresholds_skips_unrepresentable_integer() -> None:
valid model overrides; it belongs to the same rejected category as NaN and
infinity and is therefore omitted from the normalized mapping.
"""
-
assert settings._normalized_model_thresholds({"huge": 10**10_000}) == {}
@@ -285,7 +280,7 @@ def __init__(self, **kwargs: Any) -> None:
def test_construct_compressor_wraps_initialization_failure() -> None:
class Broken:
- def __init__(self, **kwargs: Any) -> None:
+ def __init__(self, **_kwargs: Any) -> None:
raise ValueError("boom")
with pytest.raises(settings.SettingsError, match="initialization failed"):
@@ -334,7 +329,6 @@ def test_base_url_normalization_preserves_route_identity(
relative, credential-bearing, and malformed authorities stay untouched so
normalization never invents a different route.
"""
-
assert settings.normalize_base_url(value) == expected
@@ -346,7 +340,6 @@ def test_literal_custom_provider_precedes_the_model_endpoint() -> None:
every live base-URL guard abstain and silently disables exact budgeting on
the endpoint Hermes actually calls.
"""
-
config = {
**base_config,
"model": {
@@ -372,7 +365,6 @@ def test_load_settings_uses_effective_named_custom_route() -> None:
and an empty raw ``model.base_url`` makes both public and auxiliary
budgeting reject the configured primary route entirely.
"""
-
config = {
**base_config,
"model": {
@@ -430,7 +422,6 @@ def test_load_settings_matches_all_hermes_named_provider_selectors(
``custom`` before middleware and auxiliary builders run. Retaining a
selector literal makes both route guards reject the actual primary route.
"""
-
result = load(
config={
**base_config,
@@ -455,7 +446,6 @@ def test_named_provider_identity_preserves_repeated_spaces() -> None:
whitespace first changes that identity to ``edge-name`` and makes
incontext reject a provider that the installed runtime resolves.
"""
-
result = load(
config={
"model": {
@@ -487,7 +477,6 @@ def test_load_settings_resolves_legacy_custom_provider_route(selector: str) -> N
makes route guards miss every request. Legacy normalization deliberately
drops unsupported output-cap metadata before constructing the runtime.
"""
-
ModernCompressor.calls.clear()
result = load(
config={
@@ -545,7 +534,6 @@ def test_prefixed_custom_identity_resolves_in_both_provider_schemas(
request prefix without canonicalizing the stored identity makes a valid
live route fail plugin registration.
"""
-
result = load(
config={
"model": {
@@ -576,7 +564,6 @@ def test_disabled_modern_provider_falls_through_to_legacy_entry(
silently disables exact budgeting on the valid legacy route. Boolean,
string, and truth-value forms must follow Hermes' compatibility parser.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
config_module = types.ModuleType("hermes_cli.config")
@@ -628,7 +615,6 @@ def test_older_hermes_does_not_apply_a_future_provider_enabled_flag(
incontext scope itself to a legacy fallback while the installed agent uses
the modern endpoint, silently disabling budgeting on the live request.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "hermes_cli", hermes_cli)
@@ -667,7 +653,6 @@ def test_provider_enabled_fails_open_when_installed_helper_raises(
retain the provider as older releases did rather than silently switching
to another endpoint before the agent itself resolves the route.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
config_module = types.ModuleType("hermes_cli.config")
@@ -691,7 +676,6 @@ def test_legacy_custom_provider_lookup_returns_none_without_a_match() -> None:
built-in path; selecting the last unrelated entry would scope the vLLM
tokenizer and context window to the wrong external destination.
"""
-
assert (
settings._legacy_provider_config(
[{"name": "Unrelated", "base_url": "https://unused.invalid/v1"}],
@@ -728,7 +712,6 @@ def test_named_provider_uses_hermes_effective_endpoint(
either disables budgeting on the primary route or removes the URL guard and
permits the primary tokenizer on an unrelated custom fallback.
"""
-
model: dict[str, Any] = {
"default": "qwen-test",
"provider": "custom:local",
@@ -761,7 +744,6 @@ def test_named_provider_uses_hermes_camelcase_base_url_alias() -> None:
the shared provider label ``custom``, allowing a same-model fallback to be
budgeted with the primary tokenizer and context window.
"""
-
result = load(
config={
"model": {
@@ -791,7 +773,6 @@ def test_legacy_entry_precedes_compatibility_normalized_modern_base_url() -> Non
modern ``baseUrl`` immediately gives incontext a different endpoint guard
from the agent that will send the request.
"""
-
result = load(
config={
"model": {
@@ -830,7 +811,6 @@ def test_compatibility_provider_lookup_ignores_unusable_entries(
URLs. Treating any of those as selected would remove or corrupt endpoint
isolation for the tokenizer-backed budget.
"""
-
assert settings._compatible_modern_provider_config(providers, "edge") is None
@@ -843,7 +823,6 @@ def test_compatibility_provider_lookup_skips_version_disabled_entry(
normalized later. When the installed release supports disabling entries,
that later path must skip it too or incontext revives a route Hermes hides.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
config_module = types.ModuleType("hermes_cli.config")
@@ -875,7 +854,6 @@ def test_legacy_provider_accepts_runtime_url_placeholder() -> None:
a valid provider from incontext and make plugin startup disagree with the
agent's later resolved route.
"""
-
assert settings._valid_provider_endpoint("https://${REGION}.example/v1") == (
"https://${REGION}.example/v1"
)
@@ -918,7 +896,6 @@ def test_legacy_provider_uses_hermes_normalized_endpoint(
``url`` and ``api``. Using modern precedence stores a different route and
makes middleware reject the endpoint the running agent actually calls.
"""
-
result = load(
config={
"model": {
@@ -941,7 +918,6 @@ def test_incomplete_provider_entries_fall_through_to_usable_legacy_entry() -> No
URL. Treating either incomplete record as selected removes endpoint
isolation and hides a valid legacy route with the same durable identity.
"""
-
result = load(
config={
"model": {
@@ -978,7 +954,6 @@ def test_malformed_modern_provider_data_falls_through_to_legacy(
``custom_providers``. Failing immediately aborts plugin registration even
though the agent has already selected a valid backward-compatible route.
"""
-
result = load(
config={
"model": {
@@ -1005,7 +980,6 @@ def test_bare_custom_incomplete_modern_entry_falls_through_to_legacy() -> None:
the compatibility list. Stopping at the incomplete mapping removes the
URL scope and can apply the primary tokenizer to another custom route.
"""
-
result = load(
config={
"model": {
@@ -1032,7 +1006,6 @@ def test_provider_selector_preserves_underscores_as_identity() -> None:
the first colliding provider and store an endpoint that the live agent does
not use, causing every exact-budget route guard to miss.
"""
-
result = load(
config={
"model": {
@@ -1059,7 +1032,6 @@ def test_colon_bearing_provider_selector_matches_its_literal_key() -> None:
prefix resolves the unrelated ``edge`` entry instead, attaching the wrong
endpoint and context policy to otherwise valid requests.
"""
-
result = load(
config={
"model": {
@@ -1086,7 +1058,6 @@ def test_legacy_provider_skips_malformed_higher_precedence_url() -> None:
as route identity makes incontext reject the valid endpoint actually used
by generation and silently disables exact budgeting.
"""
-
result = load(
config={
"model": {
@@ -1116,7 +1087,6 @@ def test_load_settings_uses_live_identity_for_local_provider_alias(alias: str) -
before invoking request middleware. Keeping the configured alias would
silently disable exact budgeting even though the endpoint and model match.
"""
-
result = load(
config={
**base_config,
@@ -1161,7 +1131,6 @@ def test_load_settings_rejects_invalid_provider_route_configuration(
them would either disable dynamic budgeting or apply the primary tokenizer
and context window to a different fallback endpoint.
"""
-
with pytest.raises(settings.SettingsError, match="provider"):
load(config=config)
@@ -1173,7 +1142,6 @@ def test_load_settings_preserves_nonlocal_builtin_provider() -> None:
to ``custom``. A built-in provider without a matching profiles entry must
retain its live middleware label while still using the configured model URL.
"""
-
result = load(
config={
**base_config,
@@ -1194,7 +1162,6 @@ def test_auto_provider_with_local_endpoint_uses_live_openrouter_identity() -> No
not the literal selector ``auto``; retaining ``auto`` makes route guards
reject every otherwise matching primary request.
"""
-
result = load(
config={
**base_config,
@@ -1224,7 +1191,6 @@ def test_auto_local_route_excludes_known_cloud_hosts(
expected: bool,
) -> None:
"""Apply Hermes' auto bypass only to explicit non-cloud endpoints."""
-
assert settings._auto_uses_openai_compatible_route(base_url) is expected
@@ -1238,7 +1204,6 @@ def test_load_settings_uses_hermes_live_identity_for_builtin_alias(
middleware. Retaining the configuration spelling makes both primary and
auxiliary route guards reject the intended request before exact counting.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
auth = types.ModuleType("hermes_cli.auth")
@@ -1281,7 +1246,6 @@ def test_canonical_builtin_provider_is_not_shadowed_by_custom_entry(
both route guards reject every real primary request, silently disabling
exact budgeting.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
auth = types.ModuleType("hermes_cli.auth")
@@ -1316,7 +1280,6 @@ def test_load_settings_keeps_provider_when_hermes_alias_resolution_fails(
itself treats later provider setup as authoritative, so incontext must
preserve the normalized selector rather than disable plugin registration.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
auth = types.ModuleType("hermes_cli.auth")
@@ -1349,7 +1312,6 @@ def test_load_settings_uses_hermes_normalized_model_identity(
scoping skip exact budgeting and can evaluate model-specific compression
policy under a different identity.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
model_normalize = types.ModuleType("hermes_cli.model_normalize")
@@ -1400,7 +1362,6 @@ def test_model_normalization_remains_best_effort_like_hermes(
retain the configured model under the same conditions instead of making an
optional compatibility helper a startup dependency.
"""
-
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
model_normalize = types.ModuleType("hermes_cli.model_normalize")
@@ -1496,7 +1457,6 @@ def test_effective_threshold_delegates_to_installed_hermes_policy(
forwards model, provider, opt-out state, and the final autoraise verdict
rather than reimplementing those rules.
"""
-
agent = types.ModuleType("agent")
agent.__path__ = [] # type: ignore[attr-defined]
agent_init = types.ModuleType("agent.agent_init")
@@ -1525,9 +1485,9 @@ def resolve(
agent_init._resolve_compression_threshold = resolve # type: ignore[attr-defined]
auxiliary._compression_threshold_for_model = model_threshold # type: ignore[attr-defined]
auxiliary._is_codex_gpt54_or_gpt55 = ( # type: ignore[attr-defined]
- lambda model, provider: False
+ lambda _model, _provider: False
)
- auxiliary._is_codex_spark = lambda model, provider: True # type: ignore[attr-defined]
+ auxiliary._is_codex_spark = lambda _model, _provider: True # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "agent", agent)
monkeypatch.setitem(sys.modules, "agent.agent_init", agent_init)
monkeypatch.setitem(
@@ -1567,7 +1527,6 @@ def test_effective_threshold_uses_legacy_hermes_policy(
a ``None`` result retains the configured threshold, so incontext and the
installed compressor always derive the same boundary.
"""
-
agent = types.ModuleType("agent")
agent.__path__ = [] # type: ignore[attr-defined]
auxiliary = types.ModuleType("agent.auxiliary_client")
@@ -1609,18 +1568,17 @@ def test_effective_threshold_falls_back_when_hermes_policy_fails(
If a future compatible release raises while resolving optional model policy,
incontext must retain the validated global threshold just as Hermes does.
"""
-
agent = types.ModuleType("agent")
agent.__path__ = [] # type: ignore[attr-defined]
agent_init = types.ModuleType("agent.agent_init")
auxiliary = types.ModuleType("agent.auxiliary_client")
- def model_policy(*args: Any, **kwargs: Any) -> float:
+ def model_policy(*_args: Any, **_kwargs: Any) -> float:
if failure_stage == "threshold":
raise RuntimeError("policy unavailable")
return 0.9
- def resolver(*args: Any, **kwargs: Any) -> tuple[float, None]:
+ def resolver(*_args: Any, **_kwargs: Any) -> tuple[float, None]:
if failure_stage == "resolver":
raise RuntimeError("resolver unavailable")
return 0.9, None
@@ -1628,9 +1586,9 @@ def resolver(*args: Any, **kwargs: Any) -> tuple[float, None]:
agent_init._resolve_compression_threshold = resolver # type: ignore[attr-defined]
auxiliary._compression_threshold_for_model = model_policy # type: ignore[attr-defined]
auxiliary._is_codex_gpt54_or_gpt55 = ( # type: ignore[attr-defined]
- lambda model, provider: False
+ lambda _model, _provider: False
)
- auxiliary._is_codex_spark = lambda model, provider: False # type: ignore[attr-defined]
+ auxiliary._is_codex_spark = lambda _model, _provider: False # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "agent", agent)
monkeypatch.setitem(sys.modules, "agent.agent_init", agent_init)
monkeypatch.setitem(
@@ -1658,7 +1616,6 @@ def test_load_settings_reserves_hermes_configured_output_budget() -> None:
threshold. Dropping that value here produces a larger, fictitious window
and lets incontext budget requests beyond Hermes' actual boundary.
"""
-
ModernCompressor.calls.clear()
config = {
**base_config,
@@ -1689,7 +1646,6 @@ def test_load_settings_reserves_hermes_environment_output_budget() -> None:
override is not folded into ``load_config()``, so it must be read through a
typed skelet storage or incontext would derive a larger unsafe window.
"""
-
ModernCompressor.calls.clear()
config = {
**base_config,
@@ -1721,7 +1677,6 @@ def test_load_settings_uses_only_a_selected_provider_output_budget() -> None:
metadata on an incomplete provider block must not reserve output space in
incontext when Hermes ignores that block entirely.
"""
-
ModernCompressor.calls.clear()
config = {
**base_config,
@@ -1763,7 +1718,6 @@ def test_load_settings_reserves_provider_max_tokens_alias() -> None:
compressor for the same named provider. When both aliases remain after a
configuration migration, its modern ``max_output_tokens`` value wins.
"""
-
ModernCompressor.calls.clear()
config = {
**base_config,
@@ -1801,7 +1755,6 @@ def test_blank_hermes_max_tokens_falls_back_to_model_configuration() -> None:
``model.max_tokens``. Native skelet conversion and validation must mirror
that behavior instead of failing plugin registration.
"""
-
ModernCompressor.calls.clear()
config = {
**base_config,
@@ -1833,7 +1786,6 @@ def test_invalid_hermes_max_tokens_fails_native_environment_validation() -> None
leak a raw conversion exception or silently fall through to YAML because
Hermes itself would be unable to construct a matching token allowance.
"""
-
with pytest.raises(
settings.SettingsError,
match="max_tokens must be a positive integer or blank",
@@ -1849,7 +1801,6 @@ def test_load_settings_rejects_invalid_hermes_output_budget(value: Any) -> None:
be silently converted into a different compression threshold by incontext.
The validation mirrors the strict integer rules used for context length.
"""
-
config = {
**base_config,
"model": {**base_config["model"], "max_tokens": value},
@@ -1870,7 +1821,6 @@ def test_load_settings_ignores_provider_caps_hermes_does_not_promote(
out of ``AIAgent``. Parsing the raw value changes the reconstructed
compression boundary even though the live runtime ignores it.
"""
-
config = {
**base_config,
"model": {
@@ -1894,7 +1844,7 @@ def test_load_settings_ignores_provider_caps_hermes_does_not_promote(
def test_load_settings_explicit_window_avoids_compressor_construction() -> None:
class MustNotRun:
- def __init__(self, **kwargs: Any) -> None:
+ def __init__(self, **_kwargs: Any) -> None:
raise AssertionError("compressor should not run")
result = load(
@@ -1921,7 +1871,7 @@ def test_disabled_hermes_compression_uses_the_complete_context_window(
"""
class MustNotRun:
- def __init__(self, **kwargs: Any) -> None:
+ def __init__(self, **_kwargs: Any) -> None:
raise AssertionError("inactive compressor threshold must not be used")
result = load(
@@ -1944,7 +1894,6 @@ def test_load_settings_rejects_non_builtin_context_engine() -> None:
class would silently give request middleware an unrelated window, which can
truncate valid output or cross the active engine's real boundary.
"""
-
config = {**base_config, "context": {"engine": "lcm"}}
with pytest.raises(
@@ -1962,7 +1911,6 @@ def test_explicit_window_supports_external_context_engine_safely() -> None:
compressor is constructed, so the plugin can budget against the stated
external boundary without inventing engine-specific policy.
"""
-
result = load(
environment={"INCONTEXT_COMPRESSION_WINDOW_TOKENS": "50000"},
config={**base_config, "context": {"engine": "lcm"}},
@@ -2103,7 +2051,6 @@ def test_load_settings_ignores_non_mapping_providers_for_direct_route() -> None:
whole otherwise valid profile before it can scope budgeting to the model's
explicit custom URL.
"""
-
result = load(config={**base_config, "providers": "invalid"})
assert result.provider == "custom"
@@ -2122,7 +2069,7 @@ def test_load_settings_requires_model_name(model_name: Any) -> None:
def test_load_settings_rejects_context_mismatch() -> None:
class Mismatch(ModernCompressor):
- def __init__(self, **kwargs: Any) -> None:
+ def __init__(self, **_kwargs: Any) -> None:
self.context_length = 32_000
self.threshold_tokens = 20_000
diff --git a/tests/units/test_vllm.py b/tests/units/test_vllm.py
index 3614773..13d1dbd 100644
--- a/tests/units/test_vllm.py
+++ b/tests/units/test_vllm.py
@@ -78,7 +78,6 @@ def test_vllm_maps_responses_output_cap_to_chat_completions() -> None:
translates only that incompatible alias and preserves aliases that the
endpoint already understands.
"""
-
backend, _ = make_backend()
assert backend.output_budget_field("max_output_tokens") == "max_tokens"
@@ -111,7 +110,6 @@ def test_vllm_coerces_output_caps_like_chat_request_validation(
spellings. Dynamic budgeting must preserve exactly those caller bounds;
accepting or dropping a different spelling changes the provider request.
"""
-
backend, _ = make_backend()
assert backend.coerce_output_budget(wire_value) == expected
@@ -125,7 +123,6 @@ def test_default_user_agent_tracks_distribution_version() -> None:
prevents a copied version literal from silently identifying a newer client
as an older release after future version bumps.
"""
-
with patch.dict(
"os.environ",
{"INCONTEXT_TOKENIZER_URL": "https://inference.test/tokenize"},
@@ -166,7 +163,6 @@ def test_build_payload_honors_read_only_extra_body_overrides() -> None:
tokenizes the superseded model, messages, and tools, so exact budgeting can
use an unrelated chat template and undercount the actual prompt.
"""
-
wire_messages = ({"role": "user", "content": "wire"},)
wire_tools = ({"type": "function", "function": {"name": "wire"}},)
@@ -199,7 +195,6 @@ def test_build_payload_materializes_tuple_tools_like_openai_sdk() -> None:
rendered prompt and undercounts it. Materialization must not mutate the
caller-owned tuple that Hermes can reuse for retries.
"""
-
tools = (
{
"type": "function",
@@ -231,7 +226,6 @@ def test_build_payload_recursively_materializes_openai_wire_mappings() -> None:
the tokenizer payload even though generation reaches vLLM normally,
disabling exact counting. Copies must also preserve caller ownership.
"""
-
content_part = MappingProxyType({"type": "text", "text": "hello"})
content_parts = {"only": content_part}
message = MappingProxyType(
@@ -279,7 +273,6 @@ def test_build_payload_mirrors_every_prompt_affecting_vllm_option() -> None:
payload must forward the same options with the same precedence instead of
silently restoring ``add_generation_prompt=True``.
"""
-
request = {
"model": "qwen",
"messages": [{"role": "assistant", "content": "prefix"}],
@@ -329,7 +322,6 @@ def test_count_rejects_prompt_controls_missing_from_tokenize_schema(
both normal fields and OpenAI's authoritative ``extra_body`` override must
therefore fail before transport and let the middleware estimate safely.
"""
-
backend, opener = make_backend([response(7)])
request: dict[str, Any] = {
"model": "qwen",
@@ -358,7 +350,6 @@ def test_build_payload_applies_extra_body_to_core_chat_fields() -> None:
tool schema; an explicit empty tool list must also remain distinguishable
from an omitted field for vLLM's chat renderer.
"""
-
override_messages = [{"role": "user", "content": "override"}]
request = {
"model": "original",
@@ -388,7 +379,6 @@ def test_build_payload_reproduces_vllm_reasoning_and_rag_rendering() -> None:
make Qwen's generation prompt tens of tokens different before any content
or tool-schema growth is considered.
"""
-
request = {
"model": "qwen",
"messages": [{"role": "user", "content": "answer from context"}],
@@ -424,7 +414,6 @@ def test_build_payload_normalizes_deprecated_reasoning_content() -> None:
modern null is considered unset, while a null legacy value is simply
removed, matching the provider's null-aware validator exactly.
"""
-
legacy = MappingProxyType(
{
"role": "assistant",
@@ -494,7 +483,6 @@ def test_reasoning_normalization_preserves_unvalidated_message_shapes() -> None:
malformed scalar unchanged ensures ``/tokenize`` rejects the same value as
chat generation instead of silently manufacturing a different prompt.
"""
-
assert VllmBackend._normalize_messages("invalid") == "invalid"
@@ -505,7 +493,6 @@ def test_build_payload_preserves_explicit_thinking_override() -> None:
key is absent. An explicit value must survive even when reasoning effort
would otherwise imply the opposite setting.
"""
-
payload = VllmBackend._build_payload(
{
"model": "qwen",
@@ -529,7 +516,6 @@ def test_build_payload_forwards_invalid_template_kwargs_for_vllm_validation() ->
fail open consistently instead of tokenizing defaults for an inference
request that will later be rejected or interpreted differently.
"""
-
payload = VllmBackend._build_payload(
{
"model": "qwen",
@@ -563,7 +549,6 @@ def test_build_payload_preserves_explicit_empty_tools() -> None:
from a list to ``None`` and can select different server-side rendering
defaults, so exact tokenization must retain the caller's shape.
"""
-
payload = VllmBackend._build_payload(
{"model": "qwen", "messages": [], "tools": []},
)
@@ -590,7 +575,6 @@ def test_count_accepts_and_caches_an_empty_rendered_prompt() -> None:
back to a positive rough estimate plus margin; model context length remains
independently required to be positive.
"""
-
backend, opener = make_backend([response(0)])
request = {
"model": "qwen",
@@ -607,7 +591,6 @@ def test_count_accepts_and_caches_an_empty_rendered_prompt() -> None:
@pytest.mark.parametrize("value", [True, -1, 1.5, "0"])
def test_nonnegative_response_integer_rejects_non_counts(value: Any) -> None:
"""Keep malformed tokenizer counts outside the exact-budget contract."""
-
with pytest.raises(VllmBackend.VllmBackendError, match="invalid count"):
VllmBackend._nonnegative_response_integer({"count": value}, "count")
@@ -719,7 +702,6 @@ def test_backend_rejects_unsafe_environment(
def test_backend_accepts_http_url_with_query() -> None:
"""Accept ordinary HTTP transport components after strict validation."""
-
backend, _ = make_backend(
environment={
"INCONTEXT_TOKENIZER_URL": "http://127.0.0.1:8080/tokenize?mode=1",
@@ -735,7 +717,6 @@ def test_backend_accepts_an_injected_environment() -> None:
backend keeps an injected skelet storage by reference, its tokenizer URL
must remain read-only or cached counts could outlive an endpoint change.
"""
-
with patch.dict(
"os.environ",
{"INCONTEXT_TOKENIZER_URL": "https://injected.test/tokenize"},
@@ -766,7 +747,8 @@ def test_backend_sends_exact_request_and_caches_result() -> None:
assert http_request.get_header("Content-type") == "application/json"
assert http_request.get_header("User-agent") == "incontext-tests"
assert timeout == 3.5
- assert json.loads(http_request.data or b"") == VllmBackend._build_payload(request)
+ assert isinstance(http_request.data, bytes)
+ assert json.loads(http_request.data) == VllmBackend._build_payload(request)
def test_backend_preserves_prompt_observable_json_key_order() -> None:
@@ -777,7 +759,6 @@ def test_backend_preserves_prompt_observable_json_key_order() -> None:
from the one vLLM receives for generation. Requests with different schema
order must also occupy different cache entries instead of sharing a count.
"""
-
backend, opener = make_backend([response(10), response(11)])
first_properties = {
"z_first": {"type": "string"},
@@ -813,6 +794,7 @@ def request(properties: dict[str, Any]) -> dict[str, Any]:
assert len(opener.calls) == 2
first_wire = opener.calls[0][0].data
+ assert isinstance(first_wire, bytes)
assert (
first_wire
== json.dumps(
@@ -841,7 +823,6 @@ def test_count_honors_positive_prompt_truncation() -> None:
the raw exact count reproduces the provider-visible input size and avoids
unnecessary compression caused by budgeting from tokens vLLM discards.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -859,7 +840,6 @@ def test_count_honors_extra_body_prompt_truncation_override() -> None:
Mirroring that precedence keeps counting aligned when a caller replaces a
top-level truncation limit without mutating the request passed to Hermes.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -886,7 +866,6 @@ def test_count_honors_prompt_truncation_values_coerced_by_vllm(
returns the untruncated rendering, ignoring an accepted wire value counts
tokens generation drops and can trigger premature compression.
"""
-
backend, _ = make_backend([response(120)])
assert (
@@ -914,7 +893,6 @@ def test_count_honors_zero_prompt_truncation(
for those values invents prompt tokens generation discards and can trigger
unnecessary compression instead of exposing the full output window.
"""
-
backend, _ = make_backend([response(120)])
assert (
@@ -944,7 +922,6 @@ def test_count_does_not_invent_invalid_prompt_truncation_coercions(
coercions known to match the provider contract may reduce the tokenizer's
complete rendered count.
"""
-
backend, _ = make_backend([response(120)])
assert (
@@ -969,7 +946,6 @@ def test_count_does_not_cap_multimodal_prompt_after_media_expansion() -> None:
would over-allocate output and violate the compression window. The
provider-visible ``extra_body.messages`` override is authoritative here.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -999,7 +975,6 @@ def test_count_detects_reusable_multimodal_content_materialized_by_openai() -> N
sequences misclassifies the image as text-only and clamps the expanded
prompt, which over-allocates completion tokens.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -1029,7 +1004,6 @@ def test_payload_materializes_reusable_tool_collections() -> None:
schemas even though generation receives them, allowing an unsafe output
budget whenever those schemas cross the compression boundary.
"""
-
tools_by_name = {
"first": {"type": "function", "function": {"name": "first"}},
"second": {"type": "function", "function": {"name": "second"}},
@@ -1062,7 +1036,6 @@ def test_output_budget_limit_matches_vllm_truncation_validation(
ceiling while leaving the dynamic ``-1`` sentinel and invalid values to
provider validation.
"""
-
backend, _ = make_backend()
assert (
@@ -1087,7 +1060,6 @@ def test_multimodal_truncation_still_limits_vllm_output_budget() -> None:
must therefore keep the expanded tokenizer result while the independent
output ceiling remains active.
"""
-
backend, _ = make_backend()
request = {
"model": "qwen",
@@ -1127,7 +1099,6 @@ def test_multimodal_detection_is_conservative_for_wire_message_shapes(
the raw tokenizer count. Non-message values remain the provider's
validation concern and do not themselves imply media expansion.
"""
-
assert VllmBackend._has_multimodal_content({"messages": messages}) is expected
@@ -1143,7 +1114,6 @@ def test_count_truncates_vllm_structured_text_content_parts(
Classifying these parts as media would under-allocate completion space and
can trigger premature compression.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -1168,7 +1138,6 @@ def test_count_truncates_vllm_tool_reference_content_parts() -> None:
Treating the reference as media keeps the raw tokenizer count, causing
premature compression and a needlessly smaller output allowance.
"""
-
backend, _ = make_backend([response(120)])
request = {
"model": "qwen",
@@ -1193,7 +1162,6 @@ def test_cached_raw_count_supports_distinct_truncation_limits() -> None:
retain the raw count and apply each request's limit afterwards; caching an
already-truncated value would let the first caller poison later budgets.
"""
-
backend, opener = make_backend([response(120)])
base = {"model": "qwen", "messages": []}
@@ -1224,9 +1192,8 @@ def test_count_uses_disaggregated_decode_prompt_token_ids() -> None:
length. Renderer-only controls cannot invalidate an already final token
sequence, and token-ID changes can reuse the validation cache safely.
"""
-
backend, opener = make_backend([response(999)])
- request = {
+ request: dict[str, Any] = {
"model": "qwen",
"messages": [{"role": "user", "content": "not the decode prompt"}],
"tools": [{"type": "function", "function": {"name": "lookup"}}],
@@ -1261,7 +1228,6 @@ def test_count_rejects_invalid_disaggregated_prompt_token_ids(
unrelated output budget, so the backend must surface a contract error and
let the middleware use its conservative rough estimator.
"""
-
backend, opener = make_backend()
with pytest.raises(
@@ -1290,7 +1256,6 @@ def test_falsy_disaggregated_prompt_ids_render_messages_normally(
sentinels disables exact counting and applies the rough fallback margin to
a request generation can serve normally.
"""
-
backend, opener = make_backend([response(120)])
assert (
@@ -1314,7 +1279,6 @@ def test_count_ignores_kv_transfer_metadata_without_reused_prompt_ids() -> None:
concrete ``prompt_token_ids`` key replaces the chat prompt; unrelated
metadata must not disable the ordinary exact tokenizer path.
"""
-
backend, opener = make_backend([response(120)])
assert (
@@ -1343,7 +1307,6 @@ def test_count_uses_raw_count_for_dynamic_minus_one_prompt_truncation(
its raw P-token prompt. Skipping ``/tokenize`` would unnecessarily replace
this exact count with a margin-adjusted rough estimate.
"""
-
backend, opener = make_backend([response(60, 100)])
assert (
@@ -1408,7 +1371,6 @@ def test_clear_cache_invalidates_an_inflight_tokenizer_response() -> None:
old response, but that response must not become a cache hit after
``clear_cache``; the next caller has to observe the new tokenizer result.
"""
-
started = threading.Event()
release = threading.Event()
calls: list[int] = []
@@ -1492,7 +1454,6 @@ def test_response_contract_is_validated(
Validation must finish before cache publication so the next identical call
retries the server and cannot reuse a count from the rejected route.
"""
-
backend, opener = make_backend([payload, response(7)])
request = {"model": "qwen", "messages": []}
diff --git a/tests/units/test_workflows.py b/tests/units/test_workflows.py
index 728cfbc..2ddb1f6 100644
--- a/tests/units/test_workflows.py
+++ b/tests/units/test_workflows.py
@@ -13,7 +13,6 @@ def test_release_waits_for_every_behavioral_quality_workflow() -> None:
PyPI. The release workflow must invoke each reusable quality workflow and
make its trusted-publishing job depend on all three successful results.
"""
-
release = (WORKFLOWS / "release.yml").read_text(encoding="utf-8")
assert "uses: ./.github/workflows/lint.yml" in release
@@ -34,7 +33,6 @@ def test_release_quality_workflows_expose_reusable_entry_points() -> None:
protects the package index. Lint and unit jobs must also exercise every
interpreter promised by package metadata, including free-threaded Python.
"""
-
for name in ("lint.yml", "tests_and_coverage.yml", "hermes_e2e.yml"):
contents = (WORKFLOWS / name).read_text(encoding="utf-8")
assert " workflow_call:\n" in contents
@@ -47,6 +45,9 @@ def test_release_quality_workflows_expose_reusable_entry_points() -> None:
contents = (WORKFLOWS / name).read_text(encoding="utf-8")
assert python_matrix in contents
+ lint = (WORKFLOWS / "lint.yml").read_text(encoding="utf-8")
+ assert "mypy tests --exclude tests/typing" in lint
+
def test_distribution_workflow_imports_wheel_code_in_isolation() -> None:
"""Reject metadata-only distribution checks that accept an empty wheel.
@@ -57,7 +58,6 @@ def test_distribution_workflow_imports_wheel_code_in_isolation() -> None:
Python, prove the module came from site-packages, and load both published
plugin entry points from the installed wheel.
"""
-
workflow = (WORKFLOWS / "tests_and_coverage.yml").read_text(encoding="utf-8")
assert "wheel-check/bin/python -I" in workflow