Implementation Plan: T5-P4-A7-WP1 Eliminate 'anthropic'.casefold() from _quota_refresh_loop and resolve_provider() from _prime_quota_cache

src/autoskillit/core/__init__.pyi

@@ -304,6 +304,7 @@ from .types import ProcessStaleError as ProcessStaleError
 from .types import PromptContractError as PromptContractError
 from .types import ProviderOutcome as ProviderOutcome
 from .types import PRState as PRState
+from .types import QuotaPolicy as QuotaPolicy
 from .types import QuotaRefreshTask as QuotaRefreshTask
 from .types import ReadingToken as ReadingToken
 from .types import ReadOnlyResolver as ReadOnlyResolver

src/autoskillit/core/types/_type_protocols_infra.py

@@ -2,13 +2,15 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Protocol, runtime_checkable
 
 __all__ = [
     "GateState",
     "BackgroundSupervisor",
     "FleetLock",
+    "QuotaPolicy",
     "QuotaRefreshTask",
     "TokenFactory",
     "CampaignProtector",
@@ -104,3 +106,15 @@ class CampaignProtector(Protocol):
     """
 
     def __call__(self, project_dir: Path) -> frozenset[str]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class QuotaPolicy:
+    """Typed quota capability derived from backend capabilities.
+
+    supports_quota_check is True when the active backend is
+    Anthropic-provider-capable; False when ctx.backend is None or
+    non-Anthropic.
+    """
+
+    supports_quota_check: bool = False

src/autoskillit/server/_guards.py

@@ -4,7 +4,7 @@
 
 import os
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from autoskillit.core import (
     InputContractResolver,
@@ -239,6 +239,11 @@ def _check_input_contracts(
     return None
 
 
+def _backend_supports_quota(ctx: Any) -> bool:
+    """Return True when the active backend is Anthropic-provider-capable."""
+    return ctx.backend is not None and ctx.backend.capabilities.anthropic_provider_capable
+
+
 def _provider_result(
     provider: str,
     profiles: dict[str, dict[str, str | None]],

src/autoskillit/server/_lifespan.py

@@ -64,6 +64,7 @@
     validate_plugin_cache_hooks,
 )
 from autoskillit.pipeline import create_background_task
+from autoskillit.server._guards import _backend_supports_quota
 from autoskillit.server._state import _get_ctx_or_none, deferred_initialize
 
 logger = get_logger(__name__)
@@ -253,7 +254,6 @@ async def _fleet_auto_gate_boot(ctx: Any) -> None:
         from autoskillit.server._misc import (  # circular-break
             _prime_quota_cache,
             _quota_refresh_loop,
-            resolve_provider,
         )
         from autoskillit.server.tools.tools_kitchen import _write_hook_config  # circular-break
 
@@ -269,16 +269,18 @@ async def _fleet_auto_gate_boot(ctx: Any) -> None:
     except Exception:
         logger.warning("fleet_auto_gate_boot_write_hook_config_failed", exc_info=True)
 
+    _supports_quota = _backend_supports_quota(ctx)
+
     try:
-        await _prime_quota_cache()
+        await _prime_quota_cache(supports_quota_check=_supports_quota)
     except Exception:
         logger.warning("fleet_auto_gate_boot_prime_quota_cache_failed", exc_info=True)
 
     try:
         ctx.quota_refresh_task = create_background_task(
             _quota_refresh_loop(
                 ctx.config.quota_guard,
-                provider=resolve_provider(ctx.config.providers.default_provider),
+                supports_quota_check=_supports_quota,
             ),
             label="quota_refresh_loop",
         )
@@ -341,7 +343,8 @@ async def _pre_reveal_kitchen(ctx: Any) -> None:
         _mcp.disable(tags={tag})
     register_active_kitchen(ctx.kitchen_id, os.getpid(), str(ctx.project_dir))
     _write_hook_config()
-    await _prime_quota_cache()
+    _supports_quota = _backend_supports_quota(ctx)
+    await _prime_quota_cache(supports_quota_check=_supports_quota)
     ctx.gate_infrastructure_ready = True
 
 
@@ -355,7 +358,6 @@ async def _food_truck_auto_gate_boot(ctx: Any) -> None:
     from autoskillit.server._misc import (  # circular-break
         _prime_quota_cache,
         _quota_refresh_loop,
-        resolve_provider,
     )
     from autoskillit.server.tools.tools_kitchen import _write_hook_config  # circular-break
 
@@ -401,8 +403,10 @@ async def _food_truck_auto_gate_boot(ctx: Any) -> None:
     except Exception:
         logger.warning("food_truck_auto_gate_boot_hook_config_failed", exc_info=True)
 
+    _supports_quota = _backend_supports_quota(ctx)
+
     try:
-        await _prime_quota_cache()
+        await _prime_quota_cache(supports_quota_check=_supports_quota)
     except Exception:
         logger.warning("food_truck_auto_gate_boot_quota_cache_failed", exc_info=True)
 
@@ -411,7 +415,7 @@ async def _food_truck_auto_gate_boot(ctx: Any) -> None:
             ctx.quota_refresh_task = create_background_task(
                 _quota_refresh_loop(
                     ctx.config.quota_guard,
-                    provider=resolve_provider(ctx.config.providers.default_provider),
+                    supports_quota_check=_supports_quota,
                 ),
                 label="quota_refresh_loop",
             )
@@ -514,7 +518,8 @@ async def _skill_auto_gate_boot(ctx: Any) -> None:
     try:
         from autoskillit.server._misc import _prime_quota_cache  # circular-break
 
-        await _prime_quota_cache()
+        _supports_quota = _backend_supports_quota(ctx)
+        await _prime_quota_cache(supports_quota_check=_supports_quota)
     except Exception:
         logger.warning("skill_auto_gate_boot_quota_cache_failed", exc_info=True)
 

src/autoskillit/server/_misc.py

@@ -236,30 +236,27 @@ async def resolve_repo_from_remote(cwd: str, hint: str | None = None) -> str:
     return await resolve_remote_repo(cwd, hint=hint) or ""
 
 
-def resolve_provider(default_provider: str | None) -> str:
-    """Return the effective provider name, falling back to ``"anthropic"``."""
-    return default_provider if default_provider else "anthropic"
-
-
-async def _prime_quota_cache() -> None:
+async def _prime_quota_cache(*, supports_quota_check: bool) -> None:
     """Fetch quota from the Anthropic API and write the local cache.
 
     Called at open_kitchen so the cache is primed before any run_skill hook fires.
     Fails open: a quota fetch failure must not abort kitchen open.
     """
+    if not supports_quota_check:
+        return
     from autoskillit.server._state import _get_ctx as _ctx_fn  # circular-break
 
     try:
         _ctx = _ctx_fn()
         await check_and_sleep_if_needed(
             _ctx.config.quota_guard,
-            provider=resolve_provider(_ctx.config.providers.default_provider),
+            provider="anthropic",
         )
     except Exception:
         logger.warning("quota_prime_failed", exc_info=True)
 
 
-async def _quota_refresh_loop(config: QuotaGuardConfig, *, provider: str = "anthropic") -> None:
+async def _quota_refresh_loop(config: QuotaGuardConfig, *, supports_quota_check: bool) -> None:
     """Long-running coroutine: refreshes the quota cache every cache_refresh_interval seconds.
 
     Designed to run as a background asyncio.Task for the duration of a kitchen session.
@@ -273,9 +270,10 @@ async def _quota_refresh_loop(config: QuotaGuardConfig, *, provider: str = "anth
 
     Args:
         config: QuotaGuardConfig instance.
-        provider: Provider name. Non-anthropic providers skip the refresh loop entirely.
+        supports_quota_check: When False, exits immediately — the active backend
+            does not support Anthropic quota checks.
     """
-    if provider.casefold() != "anthropic":
+    if not supports_quota_check:
         return
 
     while True:

src/autoskillit/server/tools/tools_fleet_dispatch.py

@@ -271,7 +271,6 @@ async def dispatch_food_truck(
             _refresh_quota_cache,
             check_and_sleep_if_needed,
             invalidate_cache,
-            resolve_provider,
         )
 
         parsed_checkpoint = (
@@ -357,6 +356,10 @@ async def dispatch_food_truck(
             if _preflight_err is not None:
                 return _preflight_err
 
+        _supports_quota = (
+            tool_ctx.backend is not None
+            and tool_ctx.backend.capabilities.anthropic_provider_capable
+        )
         result = await execute_dispatch(
             tool_ctx=tool_ctx,
             recipe=recipe,
@@ -373,7 +376,7 @@ async def dispatch_food_truck(
             ),
             quota_checker=lambda cfg: check_and_sleep_if_needed(
                 cfg,
-                provider=resolve_provider(tool_ctx.config.providers.default_provider),
+                provider="anthropic" if _supports_quota else "",
             ),
             quota_refresher=_refresh_quota_cache,
             cache_invalidator=invalidate_cache,

src/autoskillit/server/tools/tools_kitchen.py

@@ -42,7 +42,7 @@
 from autoskillit.fleet import FleetSemaphore
 from autoskillit.pipeline import create_background_task
 from autoskillit.server import mcp
-from autoskillit.server._guards import _require_orchestrator_exact
+from autoskillit.server._guards import _backend_supports_quota, _require_orchestrator_exact
 from autoskillit.server._misc import (
     _apply_triage_gate,
     _build_hook_diagnostic_warning,
@@ -52,7 +52,6 @@
     _prime_quota_cache,
     _quota_refresh_loop,
     resolve_log_dir,
-    resolve_provider,
     strip_ingredients_only_keys,
 )
 from autoskillit.server._notify import track_response_size
@@ -271,6 +270,7 @@ async def _open_kitchen_handler() -> str | None:
     ctx.active_recipe_steps = {}
     ctx.active_recipe_ingredients = frozenset()
     logger.info("open_kitchen", gate_state="open", kitchen_id=ctx.kitchen_id)
+    _supports_quota = _backend_supports_quota(ctx)
 
     try:
         _write_hook_config()
@@ -280,7 +280,7 @@ async def _open_kitchen_handler() -> str | None:
         return _kitchen_failure_envelope(exc, stage="write_hook_config")
 
     try:
-        await _prime_quota_cache()
+        await _prime_quota_cache(supports_quota_check=_supports_quota)
     except Exception as exc:
         ctx.gate.disable()
         logger.warning("open_kitchen_failure", stage="prime_quota_cache", exc_info=True)
@@ -292,7 +292,7 @@ async def _open_kitchen_handler() -> str | None:
         ctx.quota_refresh_task = create_background_task(
             _quota_refresh_loop(
                 ctx.config.quota_guard,
-                provider=resolve_provider(ctx.config.providers.default_provider),
+                supports_quota_check=_supports_quota,
             ),
             label="quota_refresh_loop",
         )
@@ -565,11 +565,12 @@ async def open_kitchen(
         else:
             _ctx_post = _get_ctx()
             if _ctx_post.quota_refresh_task is None:
+                _supports_quota_post = _backend_supports_quota(_ctx_post)
                 try:
                     _ctx_post.quota_refresh_task = create_background_task(
                         _quota_refresh_loop(
                             _ctx_post.config.quota_guard,
-                            provider=resolve_provider(_ctx_post.config.providers.default_provider),
+                            supports_quota_check=_supports_quota_post,
                         ),
                         label="quota_refresh_loop",
                     )

tests/arch/test_quota_capability_isolation.py

@@ -1,8 +1,8 @@
 """AST guard: quota modules must not reference BackendCapabilities fields.
 
-Quota provider gates must use config-string comparison (resolve_provider),
-never backend capability fields. This prevents accidental coupling between
-the quota system and the backend capabilities system.
+Quota provider gates use a supports_quota_check bool passed by callers,
+never backend capability fields directly. This prevents accidental coupling
+between the quota system and the backend capabilities system.
 """
 
 from __future__ import annotations
@@ -61,7 +61,7 @@ def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
 
 
 def test_quota_modules_do_not_reference_backend_capabilities() -> None:
-    """Quota code paths must gate on provider string, never BackendCapabilities fields."""
+    """Quota code paths must gate on supports_quota_check bool, never BackendCapabilities fields."""  # noqa: E501
     from autoskillit.core import paths
 
     src_root = paths.pkg_root()
@@ -81,6 +81,6 @@ def test_quota_modules_do_not_reference_backend_capabilities() -> None:
 
     assert not violations, (
         "Quota modules must not reference BackendCapabilities or its fields "
-        "(use resolve_provider config string instead):\n"
+        "(quota capability is passed as supports_quota_check bool by callers):\n"
         + "\n".join(f"  {v}" for v in sorted(violations))
     )

tests/arch/test_subpackage_isolation.py

@@ -962,7 +962,7 @@ def test_data_directories_are_not_python_packages() -> None:
         "co-located with the execution engine that calls them",
     ),
     "tools_kitchen.py": (
-        1275,
+        1280,
         "REQ-CNST-010-E7: kitchen tool handlers — open_kitchen and lock_ingredients require "
         "inline validation helpers (_check_override_keys, _build_ingredient_key_suggestions) "
         "for ingredient key validation; splitting would cross import-layer boundaries; "
@@ -974,7 +974,8 @@ def test_data_directories_are_not_python_packages() -> None:
         "quota_refresh_loop deferred start, and all four gate.disable() rollback resets; "
         "_dispatch_infeasible_response helper for DOA pipeline refusal"
         "; capability admission control dispatch_feasible gating on deferred-recall and "
-        "get_recipe paths (+22 net lines)",
+        "get_recipe paths (+22 net lines)"
+        "; supports_quota_check bool at call sites replaces resolve_provider (+3 net lines)",
     ),
     "tools_execution.py": (
         1130,

tests/core/test_type_protocol_shards.py

@@ -69,6 +69,7 @@ def test_infra_shard_all():
         "QuotaRefreshTask",
         "TokenFactory",
         "CampaignProtector",
+        "QuotaPolicy",
     }
 
 

tests/infra/test_schema_version_convention.py

@@ -117,13 +117,13 @@ def _is_yaml_dump(node: ast.expr) -> bool:
     # staleness_cache.py — cache dict
     ("src/autoskillit/recipe/staleness_cache.py", 67),
     # _lifespan.py — hooks.json self-heal on startup drift (co-owned with Claude plugin system)
-    ("src/autoskillit/server/_lifespan.py", 89),
+    ("src/autoskillit/server/_lifespan.py", 90),
     # tools_kitchen.py — hook config, quota guard, git_ops_policy, ingredient locks overlay
-    ("src/autoskillit/server/tools/tools_kitchen.py", 201),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 220),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 254),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 997),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 1057),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 200),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 219),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 253),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 998),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 1058),
     # tools_pipeline_tracker.py — tracker_data dict
     ("src/autoskillit/server/tools/tools_pipeline_tracker.py", 166),
     # tools_status.py — mcp_data dict

tests/server/test_lifespan_fleet_boot.py

@@ -43,7 +43,7 @@ async def test_fleet_lifespan_auto_opens_gate(self, tool_ctx):
         assert tool_ctx.kitchen_id is not None
         assert tool_ctx.active_recipe_packs == frozenset()
         mock_write_hook_config.assert_called_once_with()
-        mock_prime_quota_cache.assert_awaited_once_with()
+        mock_prime_quota_cache.assert_awaited_once_with(supports_quota_check=True)
         mock_create_bg_task.assert_called_once()
         mock_register_kitchen.assert_called_once_with(
             tool_ctx.kitchen_id, os.getpid(), str(tool_ctx.project_dir)