0.0.5

.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

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:

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 []

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,

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,15 +101,13 @@ 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)
 
 
 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:

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: