Replace global runtime resources with explicit ownership (#1042)

## Problem

Provider, messaging, and transcription resources relied on
process-global state, leaving replacement, cancellation, and shutdown
ownership ambiguous. Separate server lifetimes could share
event-loop-bound resources or retain failed cleanup work.

## Changes

| Before | After |
| --- | --- |
| Provider clients found limiters through global singleton and scoped
registries. | Each provider instance receives and owns one explicitly
constructed limiter. |
| Messaging queues and voice pipelines relied on singleton or
module-global state. | Each platform owns its limiter and outbox, while
the application owns one injected transcriber. |
| Messaging shutdown mixed ingress, active work, delivery, and SDK
cleanup. | Application shutdown quiesces ingress, drains work, closes
delivery, then releases transcription and providers. |
| Cancelled or failed provider cleanup could be forgotten or treated as
complete. | The provider manager retains shielded generation and
unpublished-runtime cleanup until it succeeds. |
| Discord and Telegram startup tasks could outlive or poison runtime
readiness. | Platform runtimes observe long-lived tasks and retry only
independently repeatable lifecycle steps. |
| Constructor-captured security and diagnostic settings appeared
hot-applicable. | Admin marks those settings restart-required so applied
policy matches the running resource graph. |
| Lifecycle races lacked direct ownership coverage. | Deterministic
cancellation, retry, isolation, teardown, and live smoke contracts
protect the final ownership model. |

<!-- greptile_comment -->

<details open><summary><h3>Greptile Summary</h3></summary>

This PR moves runtime resources from global state into explicitly owned
application objects. The main changes are:

- Provider generations own their rate limiters and cleanup tasks.
- Messaging platforms own their limiter, outbox, ingress, and delivery
lifecycle.
- Application shutdown now runs through ordered cleanup gates.
- Voice transcription is injected as an owned runtime resource.
- Admin config marks constructor-captured settings as restart-required.
</details>

<h3>Confidence Score: 4/5</h3>

The shutdown path needs a bounded cleanup result before merging.

Cleanup steps that hang never reach the retryable incomplete-shutdown
path. ASGI shutdown can remain stuck while waiting for an external SDK,
transcriber, workflow, or provider cleanup. The retry ownership model
works only after cleanup returns or raises.

src/free_claude_code/runtime/application.py

<details><summary><h3><a href="https://www.greptile.com/trex"><img
alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="20" align="absmiddle"></a> T-Rex Logs</h3></summary>

**What T-Rex did**
- T-Rex ran the requested verification, but its local artifact
references were not uploaded.
- The validation run completed successfully with EXIT\_CODE: 0 and 62
tests passed in 3.91 seconds, using the command uv run pytest -vv
tests/runtime/test\_application\_runtime.py
tests/runtime/test\_provider\_manager.py
tests/providers/test\_provider\_runtime.py.

<a
href="https://app.greptile.com/trex/runs/14064214/artifacts"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifactsDark.svg?v=4"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"><img
alt="View all artifacts"
src="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"></picture></a>

_{<a href="https://www.greptile.com/trex"><img alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="14" align="absmiddle"></a> Ran code and verified through
T-Rex}
</details>

<details open><summary><h3>Important Files Changed</h3></summary>

| Filename | Overview |
|----------|----------|
| src/free_claude_code/runtime/application.py | Refactors shutdown into
ordered retryable cleanup gates, but cleanup awaitables can still block
shutdown forever. |
| src/free_claude_code/runtime/asgi.py | Reports incomplete runtime
shutdown when `close()` returns false. |
| src/free_claude_code/runtime/provider_manager.py | Adds owned provider
cleanup retry state and shielded generation cleanup. |

</details>

<a
href="https://app.greptile.com/api/ide/codex?prompt=IMPORTANT%3A%20Work%20in%20the%20repository%20%22alishahryar1%2Ffree-claude-code%22%20on%20the%20existing%20branch%20%22refactor%2Fruntime-owned-resources%22.%20Checkout%20that%20branch%20%E2%80%94%20do%20NOT%20create%20a%20new%20branch%20or%20open%20a%20new%20PR.%20Push%20your%20changes%20to%20%22refactor%2Fruntime-owned-resources%22.%0A%0AFix%20the%20following%201%20code%20review%20issue.%20Work%20through%20them%20one%20at%20a%20time%2C%20proposing%20concise%20fixes.%0A%0A---%0A%0A%23%23%23%20Issue%201%20of%201%0Asrc%2Ffree_claude_code%2Fruntime%2Fapplication.py%3A59%0A**Cleanup%20Await%20Blocks%20Shutdown**%0A%0AWhen%20a%20platform%20SDK%20stop%2C%20workflow%20drain%2C%20transcriber%20close%2C%20or%20provider%20cleanup%20hangs%2C%20this%20helper%20waits%20forever%20and%20never%20returns%20%60False%60.%20ASGI%20shutdown%20stays%20stuck%20in%20%60runtime.close%28%29%60%20instead%20of%20reporting%20an%20incomplete%20shutdown%2C%20so%20the%20retained%20resource%20graph%20cannot%20be%20retried%20cleanly.%0A%0A&repo=alishahryar1%2Ffree-claude-code&pr=1042&platform=github"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodexDark.svg?v=6"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodex.svg?v=6"><img
alt="Fix All in Codex"
src="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodex.svg?v=6"></picture></a>

_{Reviews (2): Last reviewed commit: ["Report incomplete runtime
shutdown to
AS..."](https://github.com/alishahryar1/free-claude-code/commit/338b2bd179c3875b15bbd52818dd04c780e5d46d)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=43454593)}

> Greptile also left **1 inline comment** on this PR.

**Context used:**

- Context used - CLAUDE.md
([source](https://app.greptile.com/alishahryar1/github/Alishahryar1/free-claude-code/-/custom-context?memory=d2fd24d8-0dec-4faf-8ee4-e085e215a2f8))

<!-- /greptile_comment -->

ARCHITECTURE.md

@@ -153,7 +153,8 @@ On final shutdown it best-effort kills registered child processes.
 
 [runtime/bootstrap.py](src/free_claude_code/runtime/bootstrap.py) is the single production composition function. The CLI
 supervisor supplies one settings snapshot and its restart callback; bootstrap
-configures logging, constructs the runtime owners, supplies
+configures logging, constructs the runtime owners and the configured voice
+transcriber, supplies
 [api/ports.py](src/free_claude_code/api/ports.py) to the pure API factory, and returns the ASGI application.
 
 [api/app.py](src/free_claude_code/api/app.py) registers routers, HTTP correlation middleware, and exception handlers around
@@ -162,8 +163,17 @@ runtime resources. `app.state.services` is the only runtime state published to
 FastAPI.
 
 [runtime/application.py](src/free_claude_code/runtime/application.py) owns process startup and shutdown, optional messaging,
-the managed CLI session manager, Admin pending state, and the injected restart
-callback. [runtime/asgi.py](src/free_claude_code/runtime/asgi.py) drives that owner from ASGI lifespan messages and preserves
+the selected transcriber, the managed CLI session manager, Admin pending state,
+and the injected restart callback. Shutdown is serialized and ordered: quiesce
+messaging ingress, cancel and drain workflow/CLI work, flush persistence, close
+delivery, close transcription, then close providers. An owner reference is
+released only after its cleanup succeeds; cancellation or failure leaves the
+incomplete graph retryable. Teardown stops at a failed dependency gate rather
+than closing resources that still-live upstream work may need, and the ASGI
+adapter reports that incomplete graph as lifespan shutdown failure. Cleanup is
+completion-driven: generic timeouts do not cancel half-closed external resources;
+the process supervisor owns any force-termination deadline.
+[runtime/asgi.py](src/free_claude_code/runtime/asgi.py) drives that owner from ASGI lifespan messages and preserves
 the concise startup-failure contract.
 
 [runtime/provider_manager.py](src/free_claude_code/runtime/provider_manager.py) is the only owner that constructs, publishes,
@@ -173,7 +183,11 @@ streaming responses release it from the response iterator's `finally` path on
 completion, failure, cancellation, or disconnect. A provider-only Admin Apply
 prepares a candidate and commits configuration before publication. New requests
 then use the candidate while old streams finish on the retired generation; its
-last lease closes it exactly once.
+last lease closes it exactly once. Final shutdown rejects new acquisition and
+replacement, waits every lease, and awaits the same manager-owned cleanup task
+even if the initiating request or lease release is cancelled. Failed generation
+or unpublished-candidate cleanup remains owned and retryable; the manager does
+not become terminal or clear its model catalog until every owned runtime closes.
 
 The manager also owns one application-lifetime provider model catalog and its
 single best-effort discovery task. The catalog survives provider replacement.
@@ -366,7 +380,14 @@ credential env var, default base URL, settings attribute names, and proxy suppor
 
 [providers/runtime/](src/free_claude_code/providers/runtime/) owns construction details for one
 closable provider generation: factory wiring, provider configuration, lazy
-provider instances, and transport cleanup. Application-level generation
+provider instances, provider-owned rate limiters, and transport cleanup. Each
+lazy provider receives a fresh `ProviderRateLimiter`; there is no process
+singleton or second limiter registry. The provider cache already guarantees one
+provider and limiter per provider ID within a generation. Retired generations
+retain their own synchronization state until request leases drain, while new
+generations and separate server instances never reuse it. Hot replacement
+therefore begins with fresh quota state; an old and new generation enforce
+independent budgets while old request leases drain. Application-level generation
 publication, request leases, model metadata, discovery orchestration, and
 configured-model validation belong to `ProviderRuntimeManager` in the runtime
 package. This separates a single generation's resources from process-lifetime
@@ -632,13 +653,14 @@ If `MESSAGING_PLATFORM` is `none`, or if the selected platform token is missing,
 the messaging bridge is skipped.
 
 `ApplicationRuntime` privately owns the selected platform runtime, the
-`MessagingWorkflow`, and its managed CLI session manager. The workflow owns
+`MessagingWorkflow`, configured `Transcriber`, and managed CLI session manager.
+The workflow owns
 conversation snapshot restoration and final persistence flush. The API sees only
 the `SessionControlPort` used to preserve `/stop` behavior.
 
 The platform factory returns a `MessagingPlatformComponents` bundle from
 [messaging/platforms/ports.py](src/free_claude_code/messaging/platforms/ports.py): a
-`MessagingRuntime` for lifecycle and inbound callbacks, an `OutboundMessenger`
+`MessagingRuntime` with separate `quiesce()` and `close()` phases, an `OutboundMessenger`
 for queued sends/edits/deletes, and an optional `VoiceCancellation` port for
 reply-scoped `/clear` during voice transcription. Workflow code depends on
 these ports, not on Telegram or Discord SDK objects.
@@ -646,7 +668,16 @@ these ports, not on Telegram or Discord SDK objects.
 Runtime adapters in
 [messaging/platforms/telegram.py](src/free_claude_code/messaging/platforms/telegram.py) and
 [messaging/platforms/discord.py](src/free_claude_code/messaging/platforms/discord.py) own SDK client
-lifecycle, event subscription, inbound handoff, and voice-note handoff. Inbound
+lifecycle, event subscription, inbound handoff, voice-note handoff, and one
+injected `MessagingRateLimiter`. The platform factory creates a fresh limiter
+for the selected runtime. `quiesce()` stops new SDK ingress and drains active
+handlers while delivery remains available; after workflow tasks settle,
+`close()` drains the outbox and limiter. Discord additionally retains, observes,
+and drains its long-lived client task and inbound-handler tasks, so an SDK exit
+after initial readiness immediately withdraws the runtime's connected state.
+Telegram retries initialization and polling as separate repeatable steps; it
+never restarts an already-running SDK application after polling bootstrap fails.
+Separate application runtimes cannot share or stop each other's queue. Inbound
 normalization lives in
 [messaging/platforms/telegram_inbound.py](src/free_claude_code/messaging/platforms/telegram_inbound.py)
 and [messaging/platforms/discord_inbound.py](src/free_claude_code/messaging/platforms/discord_inbound.py).
@@ -654,14 +685,27 @@ Outbound SDK calls live in
 [messaging/platforms/telegram_io.py](src/free_claude_code/messaging/platforms/telegram_io.py) and
 [messaging/platforms/discord_io.py](src/free_claude_code/messaging/platforms/discord_io.py). Shared
 delivery policy lives in [messaging/platforms/outbox.py](src/free_claude_code/messaging/platforms/outbox.py),
-which owns queued send/edit/list-based delete, dedup keys, limiter delegation,
-and fire-and-forget behavior. Workflow and command code request deletion of
+which requires that limiter directly and owns queued send/edit/list-based delete,
+dedup keys, and retained fire-and-forget tasks. Shutdown cancels and awaits both
+queued limiter work and arbitrary outbox work; there is no optional unthrottled
+fallback, and both owners reject admission once close begins. Workflow and command code request deletion of
 message ID lists; platform IO decides whether to use native batch deletion
 (Telegram) or internal per-message deletion (Discord).
 Shared voice-note orchestration lives in
 [messaging/platforms/voice_flow.py](src/free_claude_code/messaging/platforms/voice_flow.py), which owns
-pending voice registration, temp-file cleanup, transcription, cancellation, error
-replies, and the handoff to `IncomingMessage`.
+pending voice registration, file-size validation, temp-file cleanup,
+transcription, cancellation, error replies, and the handoff to
+`IncomingMessage`. It depends only on the consumer-owned `Transcriber` protocol
+from [messaging/voice.py](src/free_claude_code/messaging/voice.py). Bootstrap selects either the
+instance-owned local Whisper `TranscriptionService` or the provider-owned
+`NvidiaNimTranscriber`. Messaging no longer imports a provider adapter, and the
+local service retains only one lazy pipeline for its immutable runtime settings;
+caller cancellation waits for thread-backed transcription to actually exit
+before temporary files, pipelines, or credentials are released. The NIM adapter
+closes its per-call authenticated gRPC channel before that worker exits. Changing the
+credential used by an active voice backend through Admin is therefore
+restart-required, while the same provider credential remains hot-replaceable
+when voice does not use it.
 
 [messaging/workflow.py](src/free_claude_code/messaging/workflow.py) contains `MessagingWorkflow`, the
 platform-agnostic coordinator. It owns dependencies, callback wiring, stop/clear
@@ -755,6 +799,10 @@ Logging defaults are conservative:
   message logging are opt-in.
 - Messaging text, transcription previews, CLI diagnostics, and detailed
   messaging exception strings are controlled by separate diagnostic flags.
+- Process logging, server/managed-CLI authentication, and messaging diagnostics
+  are captured by their lifecycle owners at construction. Admin marks those
+  settings restart-required so an Apply cannot report success while an existing
+  runtime continues using stale security or privacy policy.
 - Values under keys that look like API keys, authorization, tokens, or secrets
   are redacted by trace helpers where structured traces are emitted.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "free-claude-code"
-version = "3.4.17"
+version = "3.4.18"
 description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
 readme = "README.md"
 requires-python = ">=3.14.0"

smoke/capabilities.py

@@ -150,9 +150,13 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         "provider_proxy_timeout_config",
         "free_claude_code.providers.runtime.ProviderRuntime",
         "provider proxy, timeout, and rate-limit settings",
-        "provider client and scoped limiter config",
+        "provider client and instance-owned limiter config",
         "provider construction failure",
-        ("tests/api/test_dependencies.py", "tests/providers/test_provider_runtime.py"),
+        (
+            "tests/api/test_dependencies.py",
+            "tests/providers/test_provider_runtime.py",
+            "tests/providers/test_provider_rate_limit.py",
+        ),
     ),
     CapabilityContract(
         "provider_routing",
@@ -271,7 +275,7 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         "provider_runtime",
         "rate_limit_and_disconnect",
         "smart_rate_limiting",
-        "free_claude_code.providers.rate_limit.GlobalRateLimiter",
+        "free_claude_code.providers.rate_limit.ProviderRateLimiter",
         "concurrent provider requests and 429/disconnect failures",
         "proactive throttle, retry, cleanup",
         "mapped provider error or smoke skip for upstream disconnect",
@@ -357,6 +361,9 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         (
             "tests/messaging/test_discord_platform.py",
             "tests/messaging/test_telegram.py",
+            "tests/messaging/test_limiter.py",
+            "tests/messaging/test_platform_outbox.py",
+            "tests/runtime/test_application_runtime.py",
         ),
         ("test_telegram_bot_api_permissions", "test_discord_bot_api_permissions"),
     ),
@@ -410,13 +417,15 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         "voice",
         "voice_transcription",
         "voice_notes",
-        "free_claude_code.messaging.voice.VoiceTranscriptionService",
+        "free_claude_code.messaging.voice.Transcriber",
         "Discord/Telegram audio file and voice backend settings",
         "transcribed prompt routed to handler",
         "missing optional extra or backend error shown to user",
         (
             "tests/messaging/test_voice_handlers.py",
             "tests/messaging/test_transcription.py",
+            "tests/messaging/test_transcription_nim.py",
+            "tests/messaging/test_platform_voice_flow.py",
         ),
         ("test_voice_transcription_backend_when_explicitly_enabled",),
     ),

smoke/features.py

@@ -214,6 +214,9 @@ FEATURE_INVENTORY: tuple[FeatureCoverage, ...] = (
         (
             "tests/messaging/test_discord_platform.py",
             "tests/messaging/test_telegram.py",
+            "tests/messaging/test_limiter.py",
+            "tests/messaging/test_platform_outbox.py",
+            "tests/runtime/test_application_runtime.py",
         ),
         (
             "test_telegram_bot_api_permissions",

smoke/lib/e2e.py

@@ -323,7 +323,10 @@ class FakePlatform:
     async def start(self) -> None:
         return None
 
-    async def stop(self) -> None:
+    async def quiesce(self) -> None:
+        return None
+
+    async def close(self) -> None:
         for task in self._tasks:
             if not task.done():
                 task.cancel()

smoke/prereq/test_voice_prereq_live.py

@@ -5,13 +5,16 @@ from pathlib import Path
 
 import pytest
 
-from free_claude_code.messaging.transcription import transcribe_audio
+from free_claude_code.messaging.transcription import TranscriptionService
+from free_claude_code.messaging.voice import Transcriber
+from free_claude_code.providers.nvidia_nim.voice import NvidiaNimTranscriber
 from smoke.lib.config import SmokeConfig
 
 pytestmark = [pytest.mark.live, pytest.mark.smoke_target("voice")]
 
 
-def test_voice_transcription_backend_when_explicitly_enabled(
+@pytest.mark.asyncio
+async def test_voice_transcription_backend_when_explicitly_enabled(
     smoke_config: SmokeConfig, tmp_path: Path
 ) -> None:
     if not smoke_config.settings.voice_note_enabled:
@@ -21,16 +24,24 @@ def test_voice_transcription_backend_when_explicitly_enabled(
 
     wav_path = tmp_path / "smoke-tone.wav"
     _write_tone_wav(wav_path)
+    transcriber: Transcriber
+    if smoke_config.settings.whisper_device == "nvidia_nim":
+        transcriber = NvidiaNimTranscriber(
+            model=smoke_config.settings.whisper_model,
+            api_key=smoke_config.settings.nvidia_nim_api_key,
+        )
+    else:
+        transcriber = TranscriptionService(
+            model=smoke_config.settings.whisper_model,
+            device=smoke_config.settings.whisper_device,
+            huggingface_api_key=smoke_config.settings.huggingface_api_key,
+        )
     try:
-        t_kw: dict[str, str] = {
-            "whisper_model": smoke_config.settings.whisper_model,
-            "whisper_device": smoke_config.settings.whisper_device,
-        }
-        if smoke_config.settings.whisper_device == "nvidia_nim":
-            t_kw["nvidia_nim_api_key"] = smoke_config.settings.nvidia_nim_api_key
-        text = transcribe_audio(wav_path, "audio/wav", **t_kw)
+        text = await transcriber.transcribe(wav_path)
     except ImportError as exc:
         pytest.skip(str(exc))
+    finally:
+        await transcriber.close()
     assert isinstance(text, str)
     assert text.strip()
 

smoke/product/test_voice_product_live.py

@@ -3,7 +3,8 @@ from pathlib import Path
 
 import pytest
 
-from free_claude_code.messaging.transcription import transcribe_audio
+from free_claude_code.messaging.transcription import TranscriptionService
+from free_claude_code.providers.nvidia_nim.voice import NvidiaNimTranscriber
 from smoke.lib.config import SmokeConfig
 from smoke.lib.e2e import VoiceFixtureDriver
 
@@ -11,7 +12,10 @@ pytestmark = [pytest.mark.live]
 
 
 @pytest.mark.smoke_target("voice")
-def test_voice_local_backend_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> None:
+@pytest.mark.asyncio
+async def test_voice_local_backend_e2e(
+    smoke_config: SmokeConfig, tmp_path: Path
+) -> None:
     if not smoke_config.settings.voice_note_enabled:
         pytest.skip("missing_env: VOICE_NOTE_ENABLED is false")
     if os.getenv("FCC_SMOKE_RUN_VOICE") != "1":
@@ -21,22 +25,25 @@ def test_voice_local_backend_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> N
 
     wav_path = tmp_path / "voice-local-product.wav"
     VoiceFixtureDriver.write_tone_wav(wav_path)
+    transcriber = TranscriptionService(
+        model=smoke_config.settings.whisper_model,
+        device=smoke_config.settings.whisper_device,
+        huggingface_api_key=smoke_config.settings.huggingface_api_key,
+    )
     try:
-        text = transcribe_audio(
-            wav_path,
-            "audio/wav",
-            whisper_model=smoke_config.settings.whisper_model,
-            whisper_device=smoke_config.settings.whisper_device,
-        )
+        text = await transcriber.transcribe(wav_path)
     except ImportError as exc:
         pytest.skip(f"missing_env: {exc}")
+    finally:
+        await transcriber.close()
 
     assert isinstance(text, str)
     assert text.strip()
 
 
 @pytest.mark.smoke_target("voice")
-def test_voice_nim_backend_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> None:
+@pytest.mark.asyncio
+async def test_voice_nim_backend_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> None:
     if not smoke_config.settings.voice_note_enabled:
         pytest.skip("missing_env: VOICE_NOTE_ENABLED is false")
     if os.getenv("FCC_SMOKE_RUN_VOICE") != "1":
@@ -48,13 +55,14 @@ def test_voice_nim_backend_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> Non
 
     wav_path = tmp_path / "voice-nim-product.wav"
     VoiceFixtureDriver.write_tone_wav(wav_path)
-    text = transcribe_audio(
-        wav_path,
-        "audio/wav",
-        whisper_model=smoke_config.settings.whisper_model,
-        whisper_device="nvidia_nim",
-        nvidia_nim_api_key=smoke_config.settings.nvidia_nim_api_key,
+    transcriber = NvidiaNimTranscriber(
+        model=smoke_config.settings.whisper_model,
+        api_key=smoke_config.settings.nvidia_nim_api_key,
     )
+    try:
+        text = await transcriber.transcribe(wav_path)
+    finally:
+        await transcriber.close()
 
     assert isinstance(text, str)
     assert text.strip()

src/free_claude_code/config/admin/manifest.py

@@ -168,6 +168,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="anthropic_auth_token",
         default="freecc",
         secret=True,
+        restart_required=True,
         description="Protects Claude/API access. It is not admin-page login.",
     ),
     ConfigFieldSpec(
@@ -424,6 +425,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="debug_platform_edits",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "DEBUG_SUBAGENT_STACK",
@@ -433,6 +435,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="debug_subagent_stack",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "LOG_RAW_API_PAYLOADS",
@@ -442,6 +445,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="log_raw_api_payloads",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "LOG_RAW_SSE_EVENTS",
@@ -460,6 +464,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="log_api_error_tracebacks",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "LOG_RAW_MESSAGING_CONTENT",
@@ -469,6 +474,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="log_raw_messaging_content",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "LOG_RAW_CLI_DIAGNOSTICS",
@@ -478,6 +484,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="log_raw_cli_diagnostics",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "LOG_MESSAGING_ERROR_DETAILS",
@@ -487,6 +494,7 @@ _NON_PROVIDER_FIELDS: tuple[ConfigFieldSpec, ...] = (
         settings_attr="log_messaging_error_details",
         default="false",
         advanced=True,
+        restart_required=True,
     ),
     ConfigFieldSpec(
         "FCC_SMOKE_MODEL_NVIDIA_NIM",

src/free_claude_code/config/admin/persistence.py

@@ -106,14 +106,25 @@ def validate_updates(updates: Mapping[str, Any]) -> dict[str, Any]:
     return prepare_admin_update(updates).validation_response()
 
 
-def changed_pending_fields(updates: Mapping[str, Any]) -> list[str]:
+def changed_pending_fields(
+    updates: Mapping[str, Any],
+    *,
+    settings: Settings,
+) -> list[str]:
     """Return changed fields that require manual runtime action."""
 
     state = load_value_state()
     pending: list[str] = []
     for key, value in updates.items():
         field = FIELD_BY_KEY.get(key)
-        if field is None or not (field.restart_required or field.session_sensitive):
+        if field is None or is_locked_source(state[key]["source"]):
+            continue
+        if field.secret and value == MASKED_SECRET:
+            continue
+        requires_restart = field.restart_required or field.session_sensitive
+        if not requires_restart:
+            requires_restart = _active_voice_credential(settings) == key
+        if not requires_restart:
             continue
         if normalize_for_env(value) == str(state[key]["value"]):
             continue
@@ -121,6 +132,14 @@ def changed_pending_fields(updates: Mapping[str, Any]) -> list[str]:
     return pending
 
 
+def _active_voice_credential(settings: Settings) -> str | None:
+    if not settings.voice_note_enabled:
+        return None
+    if settings.whisper_device == "nvidia_nim":
+        return "NVIDIA_NIM_API_KEY"
+    return "HUGGINGFACE_API_KEY"
+
+
 def prepare_admin_update(updates: Mapping[str, Any]) -> PreparedAdminUpdate:
     """Validate an update and construct its prospective Settings snapshot."""
 
@@ -128,7 +147,9 @@ def prepare_admin_update(updates: Mapping[str, Any]) -> PreparedAdminUpdate:
     effective_values = effective_values_for_validation(target_values)
     settings, errors = settings_from_values(effective_values)
     pending_fields = (
-        tuple(changed_pending_fields(updates)) if settings is not None else ()
+        tuple(changed_pending_fields(updates, settings=settings))
+        if settings is not None
+        else ()
     )
     return PreparedAdminUpdate(
         target_values=target_values,

src/free_claude_code/messaging/limiter.py

@@ -1,9 +1,4 @@
-"""
-Global Rate Limiter for Messaging Platforms.
-
-Centralizes outgoing message requests and ensures compliance with rate limits
-using a strict sliding window algorithm and a task queue.
-"""
+"""Runtime-owned queued delivery for one messaging platform."""
 
 import asyncio
 from collections import deque
@@ -12,7 +7,6 @@ from typing import Any
 
 from loguru import logger
 
-from free_claude_code.config.settings import get_settings
 from free_claude_code.core.rate_limit import (
     StrictSlidingWindowLimiter as SlidingWindowLimiter,
 )
@@ -22,43 +16,21 @@ from .safe_diagnostics import format_exception_for_log
 
 class MessagingRateLimiter:
     """
-    A thread-safe global rate limiter for messaging.
+    Rate limiter and compacting work queue for one messaging runtime.
 
     Uses a custom queue with task compaction (deduplication) to ensure
     only the latest version of a message update is processed.
     """
 
-    _instance: MessagingRateLimiter | None = None
-    _lock = asyncio.Lock()
-
-    def __new__(cls, *args, **kwargs):
-        return super().__new__(cls)
-
-    @classmethod
-    async def get_instance(
-        cls,
+    def __init__(
+        self,
         *,
-        rate_limit: int = 1,
-        rate_window: float = 1.0,
-    ) -> MessagingRateLimiter:
-        """Get the singleton instance of the limiter.
-
-        ``rate_limit`` and ``rate_window`` apply only when the singleton is first
-        created. Call :meth:`shutdown_instance` before changing parameters.
-        """
-        async with cls._lock:
-            if cls._instance is None:
-                cls._instance = cls(rate_limit=rate_limit, rate_window=rate_window)
-                # Start the background worker (tracked for graceful shutdown).
-                cls._instance._start_worker()
-        return cls._instance
-
-    def __init__(self, *, rate_limit: int, rate_window: float) -> None:
-        # Prevent double initialization in singleton
-        if hasattr(self, "_initialized"):
-            return
-
+        rate_limit: int,
+        rate_window: float,
+        log_error_details: bool = False,
+    ) -> None:
         self.limiter = SlidingWindowLimiter(rate_limit, rate_window)
+        self._log_error_details = log_error_details
         # Custom queue state - using deque for O(1) popleft
         self._queue_list: deque[str] = deque()  # Deque of dedup_keys in order
         self._queue_map: dict[
@@ -66,25 +38,27 @@ class MessagingRateLimiter:
         ] = {}
         self._condition = asyncio.Condition()
         self._shutdown = asyncio.Event()
-        self._worker_task: asyncio.Task | None = None
-
-        self._initialized = True
+        self._worker_task: asyncio.Task[None] | None = None
+        self._background_tasks: set[asyncio.Task[None]] = set()
+        self._active_futures: list[asyncio.Future[Any]] = []
+        self._closed = False
         self._paused_until = 0
 
         logger.info(
             f"MessagingRateLimiter initialized ({rate_limit} req / {rate_window}s with Task Compaction)"
         )
 
-    def _start_worker(self) -> None:
-        """Ensure the worker task exists."""
+    def start(self) -> None:
+        """Start the owned worker on the current event loop."""
+        if self._closed:
+            raise RuntimeError("Messaging rate limiter is closed.")
         if self._worker_task and not self._worker_task.done():
             return
-        # Named task helps debugging shutdown hangs.
         self._worker_task = asyncio.create_task(
             self._worker(), name="msg-limiter-worker"
         )
 
-    async def _worker(self):
+    async def _worker(self) -> None:
         """Background worker that processes queued messaging tasks."""
         logger.info("MessagingRateLimiter worker started")
         while not self._shutdown.is_set():
@@ -99,6 +73,7 @@ class MessagingRateLimiter:
 
                     dedup_key = self._queue_list.popleft()
                     func, futures = self._queue_map.pop(dedup_key)
+                    self._active_futures = futures
 
                 # Check for manual pause (FloodWait)
                 now = asyncio.get_event_loop().time()
@@ -116,6 +91,19 @@ class MessagingRateLimiter:
                         for f in futures:
                             if not f.done():
                                 f.set_result(result)
+                    except asyncio.CancelledError:
+                        for f in futures:
+                            if not f.done():
+                                f.cancel()
+                        worker = asyncio.current_task()
+                        if self._shutdown.is_set() or (
+                            worker is not None and worker.cancelling()
+                        ):
+                            raise
+                        logger.debug(
+                            "Messaging operation cancelled for key {}; worker remains active",
+                            dedup_key,
+                        )
                     except Exception as e:
                         # Report error to all futures and log it
                         for f in futures:
@@ -148,17 +136,26 @@ class MessagingRateLimiter:
                                 asyncio.get_event_loop().time() + wait_secs
                             )
                         else:
-                            d = get_settings().log_messaging_error_details
                             logger.error(
                                 "Error in limiter worker for key {}: {}",
                                 dedup_key,
-                                format_exception_for_log(e, log_full_message=d),
+                                format_exception_for_log(
+                                    e,
+                                    log_full_message=self._log_error_details,
+                                ),
                             )
+                    finally:
+                        self._active_futures = []
             except asyncio.CancelledError:
-                break
+                for future in self._active_futures:
+                    if not future.done():
+                        future.cancel()
+                self._active_futures = []
+                if self._shutdown.is_set():
+                    break
+                raise
             except Exception as e:
-                d = get_settings().log_messaging_error_details
-                if d:
+                if self._log_error_details:
                     logger.error(
                         "MessagingRateLimiter worker critical error: {}",
                         e,
@@ -171,53 +168,84 @@ class MessagingRateLimiter:
                     )
                 await asyncio.sleep(1)
 
-    async def shutdown(self, timeout: float = 2.0) -> None:
-        """Stop the background worker so process shutdown doesn't hang."""
+    async def shutdown(self, timeout: float | None = None) -> None:
+        """Cancel queued work and stop every task owned by this limiter."""
+        self._closed = True
         self._shutdown.set()
-        try:
-            async with self._condition:
-                self._condition.notify_all()
-        except Exception:
-            # Best-effort: condition may be bound to a closing loop.
-            pass
-
+        async with self._condition:
+            queued_futures = [
+                future
+                for _func, futures in self._queue_map.values()
+                for future in futures
+            ]
+            self._queue_list.clear()
+            self._queue_map.clear()
+            for future in queued_futures:
+                if not future.done():
+                    future.cancel()
+            for future in self._active_futures:
+                if not future.done():
+                    future.cancel()
+            self._condition.notify_all()
+
+        cancellation: asyncio.CancelledError | None = None
+        timeout_error: TimeoutError | None = None
         task = self._worker_task
-        if not task or task.done():
-            self._worker_task = None
-            return
-
-        task.cancel()
-        try:
-            await asyncio.wait_for(task, timeout=timeout)
-        except TimeoutError:
-            logger.warning("MessagingRateLimiter worker did not stop before timeout")
-        except asyncio.CancelledError:
-            pass
-        except Exception as e:
-            d = get_settings().log_messaging_error_details
-            logger.debug(
-                "MessagingRateLimiter worker shutdown error: {}",
-                format_exception_for_log(e, log_full_message=d),
-            )
-        finally:
+        if task and not task.done():
+            task.cancel()
+            try:
+                drain = asyncio.gather(task, return_exceptions=True)
+                if timeout is None:
+                    await drain
+                else:
+                    await asyncio.wait_for(drain, timeout=timeout)
+            except TimeoutError as exc:
+                timeout_error = exc
+            except asyncio.CancelledError as exc:
+                cancellation = exc
+        if task is None or task.done():
             self._worker_task = None
 
-    @classmethod
-    async def shutdown_instance(cls, timeout: float = 2.0) -> None:
-        """Shutdown and clear the singleton instance (safe to call multiple times)."""
-        inst = cls._instance
-        if not inst:
-            return
-        try:
-            await inst.shutdown(timeout=timeout)
-        finally:
-            cls._instance = None
+        background_tasks = tuple(self._background_tasks)
+        for background_task in background_tasks:
+            background_task.cancel()
+        if background_tasks:
+            try:
+                await asyncio.gather(*background_tasks, return_exceptions=True)
+            except asyncio.CancelledError as exc:
+                cancellation = exc
+        self._background_tasks.difference_update(
+            task for task in background_tasks if task.done()
+        )
 
-    async def _enqueue_internal(self, func, future, dedup_key, front=False):
+        if cancellation is not None:
+            raise cancellation
+        if timeout_error is not None:
+            raise TimeoutError(
+                "MessagingRateLimiter worker did not stop before timeout"
+            ) from timeout_error
+
+    async def _enqueue_internal(
+        self,
+        func: Callable[[], Awaitable[Any]],
+        future: asyncio.Future[Any],
+        dedup_key: str,
+        front: bool = False,
+    ) -> None:
         await self._enqueue_internal_multi(func, [future], dedup_key, front)
 
-    async def _enqueue_internal_multi(self, func, futures, dedup_key, front=False):
+    async def _enqueue_internal_multi(
+        self,
+        func: Callable[[], Awaitable[Any]],
+        futures: list[asyncio.Future[Any]],
+        dedup_key: str,
+        front: bool = False,
+    ) -> None:
         async with self._condition:
+            if self._closed:
+                raise RuntimeError("Messaging rate limiter is closed.")
+            if self._worker_task is None or self._worker_task.done():
+                raise RuntimeError("Messaging rate limiter has not been started.")
             if dedup_key in self._queue_map:
                 # Compaction: Update existing task with new func, append new futures
                 _old_func, old_futures = self._queue_map[dedup_key]
@@ -241,28 +269,33 @@ class MessagingRateLimiter:
         Enqueue a messaging task and return its future result.
         If dedup_key is provided, subsequent tasks with the same key will replace this one.
         """
+        self._require_running()
         if dedup_key is None:
             # Unique key to avoid deduplication
-            dedup_key = f"task_{id(func)}_{asyncio.get_event_loop().time()}"
+            dedup_key = f"task_{id(func)}_{asyncio.get_running_loop().time()}"
 
-        future = asyncio.get_event_loop().create_future()
-        await self._enqueue_internal(func, future, dedup_key)
+        future = asyncio.get_running_loop().create_future()
+        try:
+            await self._enqueue_internal(func, future, dedup_key)
+        except BaseException:
+            future.cancel()
+            raise
         return await future
 
     def fire_and_forget(
         self, func: Callable[[], Awaitable[Any]], dedup_key: str | None = None
-    ):
+    ) -> None:
         """Enqueue a task without waiting for the result."""
+        self._require_running()
         if dedup_key is None:
-            dedup_key = f"task_{id(func)}_{asyncio.get_event_loop().time()}"
+            dedup_key = f"task_{id(func)}_{asyncio.get_running_loop().time()}"
 
-        future = asyncio.get_event_loop().create_future()
-
-        async def _wrapped():
+        async def _wrapped() -> None:
             max_retries = 2
             for attempt in range(max_retries + 1):
                 try:
-                    return await self.enqueue(func, dedup_key)
+                    await self.enqueue(func, dedup_key)
+                    return
                 except Exception as e:
                     error_msg = str(e).lower()
                     # Only retry transient connectivity issues that might have slipped through
@@ -271,8 +304,7 @@ class MessagingRateLimiter:
                         x in error_msg for x in ["connect", "timeout", "broken"]
                     ):
                         wait = 2**attempt
-                        d = get_settings().log_messaging_error_details
-                        if d:
+                        if self._log_error_details:
                             logger.warning(
                                 "Limiter fire_and_forget transient error (attempt {}): {}. Retrying in {}s...",
                                 attempt + 1,
@@ -289,14 +321,22 @@ class MessagingRateLimiter:
                         await asyncio.sleep(wait)
                         continue
 
-                    d = get_settings().log_messaging_error_details
                     logger.error(
                         "Final error in fire_and_forget for key {}: {}",
                         dedup_key,
-                        format_exception_for_log(e, log_full_message=d),
+                        format_exception_for_log(
+                            e,
+                            log_full_message=self._log_error_details,
+                        ),
                     )
-                    if not future.done():
-                        future.set_exception(e)
                     break
 
-        _ = asyncio.create_task(_wrapped())
+        task = asyncio.create_task(_wrapped(), name=f"msg-limiter:{dedup_key}")
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+
+    def _require_running(self) -> None:
+        if self._closed:
+            raise RuntimeError("Messaging rate limiter is closed.")
+        if self._worker_task is None or self._worker_task.done():
+            raise RuntimeError("Messaging rate limiter has not been started.")

src/free_claude_code/messaging/platforms/discord.py

@@ -9,8 +9,10 @@ from loguru import logger
 
 from free_claude_code.core.anthropic import format_user_error_preview
 
+from ..limiter import MessagingRateLimiter
 from ..models import IncomingMessage
 from ..rendering.discord_markdown import format_status_discord
+from ..voice import Transcriber
 from .discord_inbound import (
     discord_text_message_from_event,
     discord_voice_request_from_event,
@@ -55,8 +57,7 @@ if DISCORD_AVAILABLE and _discord_module is not None:
             self._runtime = runtime
 
         async def on_ready(self) -> None:
-            self._runtime._connected = True
-            logger.info("Discord platform connected")
+            self._runtime._mark_connected()
 
         async def on_message(self, message: Any) -> None:
             await self._runtime._handle_client_message(message)
@@ -74,13 +75,8 @@ class DiscordRuntime:
         bot_token: str | None = None,
         allowed_channel_ids: str | None = None,
         *,
-        voice_note_enabled: bool = True,
-        whisper_model: str = "base",
-        whisper_device: str = "cpu",
-        huggingface_api_key: str = "",
-        nvidia_nim_api_key: str = "",
-        messaging_rate_limit: int = 1,
-        messaging_rate_window: float = 1.0,
+        limiter: MessagingRateLimiter,
+        transcriber: Transcriber | None,
         log_raw_messaging_content: bool = False,
         log_api_error_tracebacks: bool = False,
     ) -> None:
@@ -102,30 +98,45 @@ class DiscordRuntime:
         self._client = _DiscordClient(self, intents)
         self._message_handler: InboundMessageHandler | None = None
         self._connected = False
-        self._limiter: Any | None = None
-        self._start_task: asyncio.Task | None = None
+        self._accepting_messages = False
+        self._ready = asyncio.Event()
+        self._inbound_tasks: set[asyncio.Task[Any]] = set()
+        self._limiter = limiter
+        self._start_task: asyncio.Task[None] | None = None
         self.outbound = DiscordMessenger(
             get_client=lambda: self._client,
             get_discord=_get_discord,
-            get_limiter=lambda: self._limiter,
+            limiter=limiter,
         )
         self._voice_flow = VoiceNoteFlow(
-            voice_note_enabled=voice_note_enabled,
-            whisper_model=whisper_model,
-            whisper_device=whisper_device,
-            huggingface_api_key=huggingface_api_key,
-            nvidia_nim_api_key=nvidia_nim_api_key,
+            transcriber=transcriber,
             log_raw_messaging_content=log_raw_messaging_content,
             log_api_error_tracebacks=log_api_error_tracebacks,
         )
-        self._messaging_rate_limit = messaging_rate_limit
-        self._messaging_rate_window = messaging_rate_window
         self._log_raw_messaging_content = log_raw_messaging_content
         self._log_api_error_tracebacks = log_api_error_tracebacks
 
     async def _handle_client_message(self, message: Any) -> None:
         """Adapter entry point used by the internal Discord client."""
-        await self._on_discord_message(message)
+        if not self._accepting_messages:
+            return
+        task = asyncio.current_task()
+        if task is not None:
+            self._inbound_tasks.add(task)
+        try:
+            if self._accepting_messages:
+                await self._on_discord_message(message)
+        finally:
+            if task is not None:
+                self._inbound_tasks.discard(task)
+
+    def _mark_connected(self) -> None:
+        """Publish Discord readiness while this runtime accepts ingress."""
+        if not self._accepting_messages:
+            return
+        self._connected = True
+        self._ready.set()
+        logger.info("Discord platform connected")
 
     async def cancel_pending_voice(
         self, chat_id: str, reply_id: str
@@ -185,46 +196,105 @@ class DiscordRuntime:
         if not self.bot_token:
             raise ValueError("DISCORD_BOT_TOKEN is required")
 
-        from ..limiter import MessagingRateLimiter
-
-        self._limiter = await MessagingRateLimiter.get_instance(
-            rate_limit=self._messaging_rate_limit,
-            rate_window=self._messaging_rate_window,
-        )
+        self._limiter.start()
+        self._accepting_messages = True
+        self._ready.clear()
 
         self._start_task = asyncio.create_task(
             self._client.start(self.bot_token),
             name="discord-client-start",
         )
+        self._start_task.add_done_callback(self._observe_client_exit)
+        ready_task = asyncio.create_task(
+            self._ready.wait(),
+            name="discord-client-ready",
+        )
+        try:
+            done, _pending = await asyncio.wait(
+                (self._start_task, ready_task),
+                timeout=30.0,
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if not done:
+                raise RuntimeError("Discord client failed to connect within timeout")
+            if self._start_task in done:
+                await self._start_task
+                raise RuntimeError("Discord client stopped before becoming ready")
+            if self._start_task.done():
+                await self._start_task
+                raise RuntimeError("Discord client stopped unexpectedly")
+        finally:
+            ready_task.cancel()
+            await asyncio.gather(ready_task, return_exceptions=True)
 
-        max_wait = 30
-        waited = 0.0
-        while not self._connected and waited < max_wait:
-            await asyncio.sleep(0.5)
-            waited += 0.5
+        logger.info("Discord platform started")
 
-        if not self._connected:
-            raise RuntimeError("Discord client failed to connect within timeout")
+    def _observe_client_exit(self, task: asyncio.Task[None]) -> None:
+        """Observe the long-lived Discord client task and publish lost readiness."""
+        if task.cancelled():
+            exception: BaseException | None = None
+        else:
+            exception = task.exception()
 
-        logger.info("Discord platform started")
+        was_connected = self._connected
+        if not self._accepting_messages:
+            return
 
-    async def stop(self) -> None:
-        """Stop Discord SDK resources."""
-        if self._client.is_closed():
-            self._connected = False
+        self._connected = False
+        self._ready.clear()
+        if not was_connected:
             return
 
-        await self._client.close()
-        if self._start_task and not self._start_task.done():
-            try:
-                await asyncio.wait_for(self._start_task, timeout=5.0)
-            except TimeoutError, asyncio.CancelledError:
-                self._start_task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await self._start_task
+        if exception is None:
+            logger.error("Discord client stopped unexpectedly")
+        elif self._log_api_error_tracebacks:
+            logger.error("Discord client stopped unexpectedly: {}", exception)
+        else:
+            logger.error(
+                "Discord client stopped unexpectedly: exc_type={}",
+                type(exception).__name__,
+            )
 
-        self._connected = False
-        logger.info("Discord platform stopped")
+    async def quiesce(self) -> None:
+        """Stop Discord ingress and drain active SDK handlers."""
+        self._accepting_messages = False
+        try:
+            if not self._client.is_closed():
+                await self._client.close()
+        finally:
+            try:
+                await self._drain_start_task()
+            finally:
+                try:
+                    await self._drain_inbound_tasks()
+                finally:
+                    self._connected = False
+                    self._ready.clear()
+
+    async def close(self) -> None:
+        """Close Discord delivery resources after ingress is quiescent."""
+        try:
+            await self.outbound.close()
+        finally:
+            await self._limiter.shutdown()
+            logger.info("Discord platform closed")
+
+    async def _drain_start_task(self) -> None:
+        task = self._start_task
+        if task is None:
+            return
+        if not task.done():
+            task.cancel()
+        try:
+            await asyncio.gather(task, return_exceptions=True)
+        finally:
+            if task.done() and self._start_task is task:
+                self._start_task = None
+
+    async def _drain_inbound_tasks(self) -> None:
+        tasks = tuple(self._inbound_tasks)
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
 
     def on_message(self, handler: Callable[[IncomingMessage], Awaitable[None]]) -> None:
         """Register the workflow callback for inbound messages."""

src/free_claude_code/messaging/platforms/discord_io.py

@@ -3,13 +3,13 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, cast
 
+from ..limiter import MessagingRateLimiter
 from .outbox import PlatformOutbox
 
 DISCORD_MESSAGE_LIMIT = 2000
 
 ClientGetter = Callable[[], Any]
 DiscordGetter = Callable[[], Any]
-LimiterGetter = Callable[[], Any | None]
 
 
 def truncate_discord_message(text: str, limit: int = DISCORD_MESSAGE_LIMIT) -> str:
@@ -27,12 +27,12 @@ class DiscordMessenger:
         *,
         get_client: ClientGetter,
         get_discord: DiscordGetter,
-        get_limiter: LimiterGetter,
+        limiter: MessagingRateLimiter,
     ) -> None:
         self._get_client = get_client
         self._get_discord = get_discord
         self._outbox = PlatformOutbox(
-            get_limiter=get_limiter,
+            limiter=limiter,
             send=self.send_message,
             edit=self.edit_message,
             delete_many=self.delete_messages,
@@ -161,3 +161,7 @@ class DiscordMessenger:
     def fire_and_forget(self, task: Awaitable[Any]) -> None:
         """Execute a coroutine without awaiting it."""
         self._outbox.fire_and_forget(task)
+
+    async def close(self) -> None:
+        """Cancel outstanding outbound work."""
+        await self._outbox.close()

src/free_claude_code/messaging/platforms/factory.py

@@ -4,6 +4,8 @@ from dataclasses import dataclass
 
 from loguru import logger
 
+from ..limiter import MessagingRateLimiter
+from ..voice import Transcriber
 from .ports import MessagingPlatformComponents
 
 
@@ -16,14 +18,11 @@ class MessagingPlatformOptions:
     telegram_proxy_url: str = ""
     discord_bot_token: str | None = None
     allowed_discord_channels: str | None = None
-    voice_note_enabled: bool = True
-    whisper_model: str = "base"
-    whisper_device: str = "cpu"
-    huggingface_api_key: str = ""
-    nvidia_nim_api_key: str = ""
+    transcriber: Transcriber | None = None
     messaging_rate_limit: int = 1
     messaging_rate_window: float = 1.0
     log_raw_messaging_content: bool = False
+    log_messaging_error_details: bool = False
     log_api_error_tracebacks: bool = False
 
 
@@ -45,17 +44,17 @@ def create_messaging_components(
 
         from .telegram import TelegramRuntime
 
+        limiter = MessagingRateLimiter(
+            rate_limit=opts.messaging_rate_limit,
+            rate_window=opts.messaging_rate_window,
+            log_error_details=opts.log_messaging_error_details,
+        )
         runtime = TelegramRuntime(
             bot_token=bot_token,
             allowed_user_id=opts.allowed_telegram_user_id,
             telegram_proxy_url=opts.telegram_proxy_url,
-            voice_note_enabled=opts.voice_note_enabled,
-            whisper_model=opts.whisper_model,
-            whisper_device=opts.whisper_device,
-            huggingface_api_key=opts.huggingface_api_key,
-            nvidia_nim_api_key=opts.nvidia_nim_api_key,
-            messaging_rate_limit=opts.messaging_rate_limit,
-            messaging_rate_window=opts.messaging_rate_window,
+            limiter=limiter,
+            transcriber=opts.transcriber,
             log_raw_messaging_content=opts.log_raw_messaging_content,
             log_api_error_tracebacks=opts.log_api_error_tracebacks,
         )
@@ -74,16 +73,16 @@ def create_messaging_components(
 
         from .discord import DiscordRuntime
 
+        limiter = MessagingRateLimiter(
+            rate_limit=opts.messaging_rate_limit,
+            rate_window=opts.messaging_rate_window,
+            log_error_details=opts.log_messaging_error_details,
+        )
         runtime = DiscordRuntime(
             bot_token=bot_token,
             allowed_channel_ids=opts.allowed_discord_channels,
-            voice_note_enabled=opts.voice_note_enabled,
-            whisper_model=opts.whisper_model,
-            whisper_device=opts.whisper_device,
-            huggingface_api_key=opts.huggingface_api_key,
-            nvidia_nim_api_key=opts.nvidia_nim_api_key,
-            messaging_rate_limit=opts.messaging_rate_limit,
-            messaging_rate_window=opts.messaging_rate_window,
+            limiter=limiter,
+            transcriber=opts.transcriber,
             log_raw_messaging_content=opts.log_raw_messaging_content,
             log_api_error_tracebacks=opts.log_api_error_tracebacks,
         )

src/free_claude_code/messaging/platforms/outbox.py

@@ -5,13 +5,16 @@ import hashlib
 from collections.abc import Awaitable, Callable
 from typing import Any, cast
 
+from loguru import logger
+
+from ..limiter import MessagingRateLimiter
+
 SendOperation = Callable[
     [str, str, str | None, str | None, str | None],
     Awaitable[str],
 ]
 EditOperation = Callable[[str, str, str, str | None], Awaitable[None]]
 DeleteManyOperation = Callable[[str, list[str]], Awaitable[None]]
-LimiterGetter = Callable[[], Any | None]
 
 
 class PlatformOutbox:
@@ -20,15 +23,17 @@ class PlatformOutbox:
     def __init__(
         self,
         *,
-        get_limiter: LimiterGetter,
+        limiter: MessagingRateLimiter,
         send: SendOperation,
         edit: EditOperation,
         delete_many: DeleteManyOperation,
     ) -> None:
-        self._get_limiter = get_limiter
+        self._limiter = limiter
         self._send = send
         self._edit = edit
         self._delete_many = delete_many
+        self._background_tasks: set[asyncio.Future[Any]] = set()
+        self._closed = False
 
     async def queue_send_message(
         self,
@@ -40,15 +45,7 @@ class PlatformOutbox:
         message_thread_id: str | None = None,
     ) -> str | None:
         """Queue or immediately send a platform message."""
-        limiter = self._get_limiter()
-        if limiter is None:
-            return await self._send(
-                chat_id,
-                text,
-                reply_to,
-                parse_mode,
-                message_thread_id,
-            )
+        self._require_open()
 
         async def _send() -> str:
             return await self._send(
@@ -60,9 +57,9 @@ class PlatformOutbox:
             )
 
         if fire_and_forget:
-            limiter.fire_and_forget(_send)
+            self._limiter.fire_and_forget(_send)
             return None
-        return cast(str | None, await limiter.enqueue(_send))
+        return cast(str | None, await self._limiter.enqueue(_send))
 
     async def queue_edit_message(
         self,
@@ -73,19 +70,16 @@ class PlatformOutbox:
         fire_and_forget: bool = True,
     ) -> None:
         """Queue or immediately edit a platform message."""
-        limiter = self._get_limiter()
-        if limiter is None:
-            await self._edit(chat_id, message_id, text, parse_mode)
-            return
+        self._require_open()
 
         async def _edit() -> None:
             await self._edit(chat_id, message_id, text, parse_mode)
 
         dedup_key = f"edit:{chat_id}:{message_id}"
         if fire_and_forget:
-            limiter.fire_and_forget(_edit, dedup_key=dedup_key)
+            self._limiter.fire_and_forget(_edit, dedup_key=dedup_key)
         else:
-            await limiter.enqueue(_edit, dedup_key=dedup_key)
+            await self._limiter.enqueue(_edit, dedup_key=dedup_key)
 
     async def queue_delete_messages(
         self,
@@ -94,28 +88,56 @@ class PlatformOutbox:
         fire_and_forget: bool = True,
     ) -> None:
         """Queue or immediately bulk-delete platform messages."""
+        self._require_open()
         ids_snapshot = tuple(str(message_id) for message_id in message_ids)
         if not ids_snapshot:
             return
 
-        limiter = self._get_limiter()
-        if limiter is None:
-            await self._delete_many(chat_id, list(ids_snapshot))
-            return
-
         async def _delete_many() -> None:
             await self._delete_many(chat_id, list(ids_snapshot))
 
         digest = hashlib.sha256("\x1f".join(ids_snapshot).encode()).hexdigest()[:16]
         dedup_key = f"del_bulk:{chat_id}:{digest}"
         if fire_and_forget:
-            limiter.fire_and_forget(_delete_many, dedup_key=dedup_key)
+            self._limiter.fire_and_forget(_delete_many, dedup_key=dedup_key)
         else:
-            await limiter.enqueue(_delete_many, dedup_key=dedup_key)
+            await self._limiter.enqueue(_delete_many, dedup_key=dedup_key)
 
     def fire_and_forget(self, task: Awaitable[Any]) -> None:
-        """Execute a coroutine or future without awaiting it."""
-        if asyncio.iscoroutine(task):
-            _ = asyncio.create_task(task)
-        else:
-            _ = asyncio.ensure_future(task)
+        """Run and retain arbitrary outbound work until completion or shutdown."""
+        future = asyncio.ensure_future(task)
+        if self._closed:
+            future.cancel()
+            raise RuntimeError("Platform outbox is closed.")
+        self._background_tasks.add(future)
+        future.add_done_callback(self._complete_background_task)
+
+    async def close(self) -> None:
+        """Cancel and await arbitrary outbound work owned by this outbox."""
+        if not self._closed:
+            self._closed = True
+        tasks = tuple(self._background_tasks)
+        for task in tasks:
+            task.cancel()
+        try:
+            if tasks:
+                await asyncio.gather(*tasks, return_exceptions=True)
+        finally:
+            self._background_tasks.difference_update(
+                task for task in tasks if task.done()
+            )
+
+    def _complete_background_task(self, task: asyncio.Future[Any]) -> None:
+        self._background_tasks.discard(task)
+        if task.cancelled():
+            return
+        error = task.exception()
+        if error is not None:
+            logger.error(
+                "Outbound background task failed: exc_type={}",
+                type(error).__name__,
+            )
+
+    def _require_open(self) -> None:
+        if self._closed:
+            raise RuntimeError("Platform outbox is closed.")

src/free_claude_code/messaging/platforms/ports.py

@@ -11,14 +11,16 @@ InboundMessageHandler = Callable[[IncomingMessage], Awaitable[None]]
 
 @runtime_checkable
 class MessagingRuntime(Protocol):
-    """Owns inbound SDK lifecycle for one messaging platform."""
+    """Owns ingress and delivery lifecycle for one messaging platform."""
 
     @property
     def name(self) -> str: ...
 
     async def start(self) -> None: ...
 
-    async def stop(self) -> None: ...
+    async def quiesce(self) -> None: ...
+
+    async def close(self) -> None: ...
 
     def on_message(self, handler: InboundMessageHandler) -> None: ...
 

src/free_claude_code/messaging/platforms/telegram.py

@@ -4,7 +4,6 @@ import asyncio
 import contextlib
 import os
 from collections.abc import Awaitable, Callable
-from typing import Any
 
 # Opt-in to future behavior for python-telegram-bot (retry_after as timedelta).
 os.environ["PTB_TIMEDELTA"] = "1"
@@ -13,8 +12,10 @@ from loguru import logger
 
 from free_claude_code.core.anthropic import format_user_error_preview
 
+from ..limiter import MessagingRateLimiter
 from ..models import IncomingMessage
 from ..rendering.telegram_markdown import escape_md_v2
+from ..voice import Transcriber
 from .ports import InboundMessageHandler
 from .telegram_inbound import (
     telegram_text_message_from_update,
@@ -50,13 +51,8 @@ class TelegramRuntime:
         allowed_user_id: str | None = None,
         *,
         telegram_proxy_url: str = "",
-        voice_note_enabled: bool = True,
-        whisper_model: str = "base",
-        whisper_device: str = "cpu",
-        huggingface_api_key: str = "",
-        nvidia_nim_api_key: str = "",
-        messaging_rate_limit: int = 1,
-        messaging_rate_window: float = 1.0,
+        limiter: MessagingRateLimiter,
+        transcriber: Transcriber | None,
         log_raw_messaging_content: bool = False,
         log_api_error_tracebacks: bool = False,
     ) -> None:
@@ -74,22 +70,16 @@ class TelegramRuntime:
         self._application: Application | None = None
         self._message_handler: InboundMessageHandler | None = None
         self._connected = False
-        self._limiter: Any | None = None
+        self._limiter = limiter
         self.outbound = TelegramMessenger(
             get_application=lambda: self._application,
-            get_limiter=lambda: self._limiter,
+            limiter=limiter,
         )
         self._voice_flow = VoiceNoteFlow(
-            voice_note_enabled=voice_note_enabled,
-            whisper_model=whisper_model,
-            whisper_device=whisper_device,
-            huggingface_api_key=huggingface_api_key,
-            nvidia_nim_api_key=nvidia_nim_api_key,
+            transcriber=transcriber,
             log_raw_messaging_content=log_raw_messaging_content,
             log_api_error_tracebacks=log_api_error_tracebacks,
         )
-        self._messaging_rate_limit = messaging_rate_limit
-        self._messaging_rate_window = messaging_rate_window
         self._log_raw_messaging_content = log_raw_messaging_content
         self._log_api_error_tracebacks = log_api_error_tracebacks
 
@@ -128,51 +118,31 @@ class TelegramRuntime:
                 connection_pool_size=8, connect_timeout=30.0, read_timeout=30.0
             )
             builder = Application.builder().token(self.bot_token).request(request)
-        self._application = builder.build()
+        application = builder.build()
+        self._application = application
 
-        self._application.add_handler(
+        application.add_handler(
             MessageHandler(filters.TEXT & (~filters.COMMAND), self._on_telegram_message)
         )
-        self._application.add_handler(CommandHandler("start", self._on_start_command))
-        self._application.add_handler(
+        application.add_handler(CommandHandler("start", self._on_start_command))
+        application.add_handler(
             MessageHandler(filters.COMMAND, self._on_telegram_message)
         )
-        self._application.add_handler(
-            MessageHandler(filters.VOICE, self._on_telegram_voice)
-        )
+        application.add_handler(MessageHandler(filters.VOICE, self._on_telegram_voice))
 
-        max_retries = 3
-        for attempt in range(max_retries):
-            try:
-                await self._application.initialize()
-                await self._application.start()
-                if self._application.updater:
-                    await self._application.updater.start_polling(
-                        drop_pending_updates=False
-                    )
-                self._connected = True
-                break
-            except Exception as e:
-                if attempt < max_retries - 1:
-                    wait_time = 2 * (attempt + 1)
-                    logger.warning(
-                        "Connection failed (attempt {}/{}): {}. Retrying in {}s...",
-                        attempt + 1,
-                        max_retries,
-                        e,
-                        wait_time,
-                    )
-                    await asyncio.sleep(wait_time)
-                else:
-                    logger.error("Failed to connect after {} attempts", max_retries)
-                    raise
-
-        from ..limiter import MessagingRateLimiter
-
-        self._limiter = await MessagingRateLimiter.get_instance(
-            rate_limit=self._messaging_rate_limit,
-            rate_window=self._messaging_rate_window,
+        await self._retry_connection_step(
+            application.initialize,
+            step="initialization",
         )
+        await application.start()
+        self._limiter.start()
+        updater = application.updater
+        if updater is not None:
+            await self._retry_connection_step(
+                lambda: updater.start_polling(drop_pending_updates=False),
+                step="polling",
+            )
+        self._connected = True
 
         try:
             target = self.allowed_user_id
@@ -193,15 +163,75 @@ class TelegramRuntime:
 
         logger.info("Telegram platform started (Bot API)")
 
-    async def stop(self) -> None:
-        """Stop Telegram polling and SDK resources."""
-        if self._application and self._application.updater:
-            await self._application.updater.stop()
-            await self._application.stop()
-            await self._application.shutdown()
+    async def _retry_connection_step(
+        self,
+        operation: Callable[[], Awaitable[object]],
+        *,
+        step: str,
+    ) -> None:
+        """Retry one independently repeatable Telegram connection step."""
+        max_attempts = 3
+        for attempt in range(1, max_attempts + 1):
+            try:
+                await operation()
+                return
+            except Exception as exc:
+                if attempt == max_attempts:
+                    logger.error(
+                        "Telegram {} failed after {} attempts",
+                        step,
+                        max_attempts,
+                    )
+                    raise
+                wait_time = 2 * attempt
+                if self._log_api_error_tracebacks:
+                    logger.warning(
+                        "Telegram {} failed (attempt {}/{}): {}. Retrying in {}s...",
+                        step,
+                        attempt,
+                        max_attempts,
+                        exc,
+                        wait_time,
+                    )
+                else:
+                    logger.warning(
+                        "Telegram {} failed (attempt {}/{}): exc_type={}. Retrying in {}s...",
+                        step,
+                        attempt,
+                        max_attempts,
+                        type(exc).__name__,
+                        wait_time,
+                    )
+                await asyncio.sleep(wait_time)
 
-        self._connected = False
-        logger.info("Telegram platform stopped")
+    async def quiesce(self) -> None:
+        """Stop Telegram ingress after draining active SDK handlers."""
+        application = self._application
+        updater = application.updater if application is not None else None
+        try:
+            if updater is not None and updater.running:
+                await updater.stop()
+        finally:
+            try:
+                if application is not None and application.running:
+                    await application.stop()
+            finally:
+                self._connected = False
+
+    async def close(self) -> None:
+        """Close Telegram delivery and initialized SDK resources."""
+        application = self._application
+        try:
+            await self.outbound.close()
+        finally:
+            try:
+                await self._limiter.shutdown()
+            finally:
+                try:
+                    if application is not None:
+                        await application.shutdown()
+                finally:
+                    logger.info("Telegram platform closed")
 
     def on_message(self, handler: Callable[[IncomingMessage], Awaitable[None]]) -> None:
         """Register the workflow callback for inbound messages."""

src/free_claude_code/messaging/platforms/telegram_io.py

@@ -7,6 +7,7 @@ from typing import Any
 
 from loguru import logger
 
+from ..limiter import MessagingRateLimiter
 from .outbox import PlatformOutbox
 
 TELEGRAM_DELETE_MESSAGES_BATCH_SIZE = 100
@@ -34,7 +35,6 @@ except ImportError:
     TelegramBaseError = Exception
 
 ApplicationGetter = Callable[[], Any | None]
-LimiterGetter = Callable[[], Any | None]
 
 
 class TelegramMessenger:
@@ -44,11 +44,11 @@ class TelegramMessenger:
         self,
         *,
         get_application: ApplicationGetter,
-        get_limiter: LimiterGetter,
+        limiter: MessagingRateLimiter,
     ) -> None:
         self._get_application = get_application
         self._outbox = PlatformOutbox(
-            get_limiter=get_limiter,
+            limiter=limiter,
             send=self.send_message,
             edit=self.edit_message,
             delete_many=self.delete_messages,
@@ -281,3 +281,7 @@ class TelegramMessenger:
     def fire_and_forget(self, task: Awaitable[Any]) -> None:
         """Execute a coroutine without awaiting it."""
         self._outbox.fire_and_forget(task)
+
+    async def close(self) -> None:
+        """Cancel outstanding outbound work."""
+        await self._outbox.close()

src/free_claude_code/messaging/platforms/voice_flow.py

@@ -1,5 +1,6 @@
 """Shared voice-note flow for messaging platform adapters."""
 
+import asyncio
 import contextlib
 import tempfile
 from collections.abc import Awaitable, Callable
@@ -12,9 +13,10 @@ from loguru import logger
 from free_claude_code.core.anthropic import format_user_error_preview
 
 from ..models import IncomingMessage
-from ..voice import PendingVoiceRegistry, VoiceTranscriptionService
+from ..voice import PendingVoiceRegistry, Transcriber
 
 AUDIO_EXTENSIONS = (".ogg", ".mp4", ".mp3", ".wav", ".m4a")
+MAX_AUDIO_SIZE_BYTES = 25 * 1024 * 1024
 VOICE_DISABLED_MESSAGE = "Voice notes are disabled."
 VOICE_TRANSCRIPTION_ERROR_MESSAGE = (
     "Could not transcribe voice note. Please try again or send text."
@@ -87,35 +89,25 @@ class VoiceNoteFlow:
     def __init__(
         self,
         *,
-        voice_note_enabled: bool,
-        whisper_model: str,
-        whisper_device: str,
-        huggingface_api_key: str,
-        nvidia_nim_api_key: str,
+        transcriber: Transcriber | None,
         log_raw_messaging_content: bool,
         log_api_error_tracebacks: bool,
     ) -> None:
-        self._voice_note_enabled = voice_note_enabled
-        self._whisper_model = whisper_model
-        self._whisper_device = whisper_device
+        self._transcriber = transcriber
         self._log_raw_messaging_content = log_raw_messaging_content
         self._log_api_error_tracebacks = log_api_error_tracebacks
         self._pending_voice = PendingVoiceRegistry()
-        self._voice_transcription = VoiceTranscriptionService(
-            huggingface_api_key=huggingface_api_key,
-            nvidia_nim_api_key=nvidia_nim_api_key,
-        )
 
     @property
     def is_enabled(self) -> bool:
         """Return whether voice-note handling is enabled."""
-        return self._voice_note_enabled
+        return self._transcriber is not None
 
     async def reply_if_disabled(
         self, reply_text: Callable[[str], Awaitable[None]]
     ) -> bool:
         """Reply with the disabled message when voice-note handling is disabled."""
-        if self._voice_note_enabled:
+        if self.is_enabled:
             return False
         await reply_text(VOICE_DISABLED_MESSAGE)
         return True
@@ -180,13 +172,12 @@ class VoiceNoteFlow:
 
         try:
             await request.download_to(tmp_path)
+            _validate_audio_file(tmp_path)
 
-            transcribed = await self._voice_transcription.transcribe(
-                tmp_path,
-                request.content_type,
-                whisper_model=self._whisper_model,
-                whisper_device=self._whisper_device,
-            )
+            transcriber = self._transcriber
+            if transcriber is None:
+                raise RuntimeError("Voice transcription is not configured.")
+            transcribed = await transcriber.transcribe(tmp_path)
 
             if not await self.is_voice_still_pending(
                 request.chat_id,
@@ -218,6 +209,14 @@ class VoiceNoteFlow:
             self._log_transcription(request, transcribed)
             await message_handler(incoming)
             return True
+        except asyncio.CancelledError:
+            await self._clear_failed_pending_voice(
+                request,
+                status_msg_id_text,
+                queue_delete_messages,
+                handed_off=handed_off,
+            )
+            raise
         except ValueError as e:
             await self._clear_failed_pending_voice(
                 request,
@@ -291,3 +290,13 @@ class VoiceNoteFlow:
                 request.message_id,
                 len(transcribed),
             )
+
+
+def _validate_audio_file(file_path: Path) -> None:
+    if not file_path.exists():
+        raise FileNotFoundError(f"Audio file not found: {file_path}")
+    size = file_path.stat().st_size
+    if size > MAX_AUDIO_SIZE_BYTES:
+        raise ValueError(
+            f"Audio file too large ({size} bytes). Max {MAX_AUDIO_SIZE_BYTES} bytes."
+        )

src/free_claude_code/messaging/transcription.py

@@ -1,23 +1,11 @@
-"""Voice note transcription for messaging platforms.
-
-Supports:
-- Local Whisper (cpu/cuda): Hugging Face transformers pipeline
-- NVIDIA NIM: NVIDIA NIM Whisper/Parakeet
-"""
+"""Instance-owned local Whisper transcription."""
 
+import asyncio
 from pathlib import Path
 from typing import Any
 
 from loguru import logger
 
-from free_claude_code.providers.nvidia_nim.voice import (
-    transcribe_audio_file as transcribe_nvidia_nim_audio,
-)
-
-# Max file size in bytes (25 MB)
-MAX_AUDIO_SIZE_BYTES = 25 * 1024 * 1024
-
-# Short model names -> full Hugging Face model IDs (for local Whisper)
 _MODEL_MAP: dict[str, str] = {
     "tiny": "openai/whisper-tiny",
     "base": "openai/whisper-base",
@@ -27,143 +15,118 @@ _MODEL_MAP: dict[str, str] = {
     "large-v3": "openai/whisper-large-v3",
     "large-v3-turbo": "openai/whisper-large-v3-turbo",
 }
+_WHISPER_SAMPLE_RATE = 16000
 
-# Lazy-loaded pipelines: (model_id, device, Hugging Face API key fingerprint) -> pipeline
-_pipeline_cache: dict[tuple[str, str, str], Any] = {}
-
-
-def _resolve_model_id(whisper_model: str) -> str:
-    """Resolve short name to full Hugging Face model ID."""
-    return _MODEL_MAP.get(whisper_model, whisper_model)
 
+class TranscriptionService:
+    """Own one lazily loaded local Whisper pipeline."""
 
-def _get_pipeline(model_id: str, device: str, huggingface_api_key: str = "") -> Any:
-    """Lazy-load transformers Whisper pipeline. Raises ImportError if not installed."""
-    global _pipeline_cache
-    if device not in ("cpu", "cuda"):
-        raise ValueError(f"whisper_device must be 'cpu' or 'cuda', got {device!r}")
-    resolved_token = huggingface_api_key or ""
-    cache_key = (model_id, device, resolved_token)
-    if cache_key not in _pipeline_cache:
+    def __init__(
+        self,
+        *,
+        model: str,
+        device: str,
+        huggingface_api_key: str = "",
+    ) -> None:
+        if device not in {"cpu", "cuda"}:
+            raise ValueError(
+                f"Local Whisper device must be 'cpu' or 'cuda', got {device!r}"
+            )
+        self._model_id = _MODEL_MAP.get(model, model)
+        self._device = device
+        self._huggingface_api_key = huggingface_api_key
+        self._pipeline: Any | None = None
+        self._lock = asyncio.Lock()
+        self._closed = False
+
+    async def transcribe(self, file_path: Path) -> str:
+        """Transcribe one audio file without blocking the event loop."""
+        async with self._lock:
+            if self._closed:
+                raise RuntimeError("Transcription service is closed.")
+            worker = asyncio.create_task(
+                asyncio.to_thread(self._transcribe_sync, file_path)
+            )
+            try:
+                return await asyncio.shield(worker)
+            except asyncio.CancelledError:
+                await _wait_for_thread_exit(worker)
+                raise
+
+    async def close(self) -> None:
+        """Prevent new work and release the owned model pipeline."""
+        self._closed = True
+        async with self._lock:
+            self._pipeline = None
+            self._huggingface_api_key = ""
+
+    def _transcribe_sync(self, file_path: Path) -> str:
+        pipe = self._get_pipeline()
+        audio = _load_audio(file_path)
+        result = pipe(audio, generate_kwargs={"language": "en", "task": "transcribe"})
+        text = result.get("text", "") or ""
+        if isinstance(text, list):
+            text = " ".join(text) if text else ""
+        result_text = text.strip()
+        logger.debug("Local transcription: {} chars", len(result_text))
+        return result_text or "(no speech detected)"
+
+    def _get_pipeline(self) -> Any:
+        if self._pipeline is not None:
+            return self._pipeline
         try:
             import torch
             from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline
-
-            hf_auth_token = resolved_token or None
-
-            use_cuda = device == "cuda" and torch.cuda.is_available()
-            pipe_device = "cuda:0" if use_cuda else "cpu"
-            model_dtype = torch.float16 if use_cuda else torch.float32
-
-            model = AutoModelForSpeechSeq2Seq.from_pretrained(
-                model_id,
-                dtype=model_dtype,
-                low_cpu_mem_usage=True,
-                attn_implementation="sdpa",
-                token=hf_auth_token,
-            )
-            model = model.to(pipe_device)
-            processor = AutoProcessor.from_pretrained(model_id, token=hf_auth_token)
-
-            pipe = pipeline(
-                "automatic-speech-recognition",
-                model=model,
-                tokenizer=processor.tokenizer,
-                feature_extractor=processor.feature_extractor,
-                device=pipe_device,
-            )
-            _pipeline_cache[cache_key] = pipe
-            logger.debug(
-                f"Loaded Whisper pipeline: model={model_id} device={pipe_device}"
-            )
-        except ImportError as e:
+        except ImportError as exc:
             raise ImportError(
-                "Local Whisper requires the voice_local extra. Install with: uv sync --extra voice_local"
-            ) from e
-    return _pipeline_cache[cache_key]
-
-
-def transcribe_audio(
-    file_path: Path,
-    mime_type: str,
-    *,
-    whisper_model: str = "base",
-    whisper_device: str = "cpu",
-    huggingface_api_key: str = "",
-    nvidia_nim_api_key: str = "",
-) -> str:
-    """
-    Transcribe audio file to text.
-
-    Supports:
-    - whisper_device="cpu"/"cuda": local Whisper (requires voice_local extra)
-    - whisper_device="nvidia_nim": NVIDIA NIM Whisper API (requires voice extra)
-
-    Args:
-        file_path: Path to audio file (OGG, MP3, MP4, WAV, M4A supported)
-        mime_type: MIME type of the audio (e.g. "audio/ogg")
-        whisper_model: Model ID or short name (local) or NVIDIA NIM model
-        whisper_device: "cpu" | "cuda" | "nvidia_nim"
-
-    Returns:
-        Transcribed text
-
-    Raises:
-        FileNotFoundError: If file does not exist
-        ValueError: If file too large
-        ImportError: If voice_local extra not installed (for local Whisper)
-    """
-
-    if not file_path.exists():
-        raise FileNotFoundError(f"Audio file not found: {file_path}")
-
-    size = file_path.stat().st_size
-    if size > MAX_AUDIO_SIZE_BYTES:
-        raise ValueError(
-            f"Audio file too large ({size} bytes). Max {MAX_AUDIO_SIZE_BYTES} bytes."
+                "Local Whisper requires the voice_local extra. "
+                "Install with: uv sync --extra voice_local"
+            ) from exc
+
+        token = self._huggingface_api_key or None
+        use_cuda = self._device == "cuda" and torch.cuda.is_available()
+        pipeline_device = "cuda:0" if use_cuda else "cpu"
+        model_dtype = torch.float16 if use_cuda else torch.float32
+        model = AutoModelForSpeechSeq2Seq.from_pretrained(
+            self._model_id,
+            dtype=model_dtype,
+            low_cpu_mem_usage=True,
+            attn_implementation="sdpa",
+            token=token,
         )
-
-    if whisper_device == "nvidia_nim":
-        return transcribe_nvidia_nim_audio(
-            file_path, whisper_model, api_key=nvidia_nim_api_key
+        model = model.to(pipeline_device)
+        processor = AutoProcessor.from_pretrained(self._model_id, token=token)
+        self._pipeline = pipeline(
+            "automatic-speech-recognition",
+            model=model,
+            tokenizer=processor.tokenizer,
+            feature_extractor=processor.feature_extractor,
+            device=pipeline_device,
         )
-    return _transcribe_local(
-        file_path,
-        whisper_model,
-        whisper_device,
-        huggingface_api_key=huggingface_api_key,
-    )
+        logger.debug(
+            "Loaded Whisper pipeline: model={} device={}",
+            self._model_id,
+            pipeline_device,
+        )
+        return self._pipeline
 
 
-# Whisper expects 16 kHz sample rate
-_WHISPER_SAMPLE_RATE = 16000
+async def _wait_for_thread_exit(worker: asyncio.Task[str]) -> None:
+    """Wait through repeated caller cancellation without cancelling thread work."""
+    while not worker.done():
+        try:
+            await asyncio.shield(asyncio.wait((worker,)))
+        except asyncio.CancelledError:
+            continue
+    if not worker.cancelled():
+        worker.exception()
 
 
 def _load_audio(file_path: Path) -> dict[str, Any]:
-    """Load audio file to waveform dict. No ffmpeg required."""
+    """Load an audio file into the waveform shape expected by Whisper."""
     import librosa
 
-    waveform, sr = librosa.load(str(file_path), sr=_WHISPER_SAMPLE_RATE, mono=True)
-    return {"array": waveform, "sampling_rate": sr}
-
-
-def _transcribe_local(
-    file_path: Path,
-    whisper_model: str,
-    whisper_device: str,
-    *,
-    huggingface_api_key: str = "",
-) -> str:
-    """Transcribe using transformers Whisper pipeline."""
-    model_id = _resolve_model_id(whisper_model)
-    pipe = _get_pipeline(
-        model_id, whisper_device, huggingface_api_key=huggingface_api_key
+    waveform, sample_rate = librosa.load(
+        str(file_path), sr=_WHISPER_SAMPLE_RATE, mono=True
     )
-    audio = _load_audio(file_path)
-    result = pipe(audio, generate_kwargs={"language": "en", "task": "transcribe"})
-    text = result.get("text", "") or ""
-    if isinstance(text, list):
-        text = " ".join(text) if text else ""
-    result_text = text.strip()
-    logger.debug(f"Local transcription: {len(result_text)} chars")
-    return result_text or "(no speech detected)"
+    return {"array": waveform, "sampling_rate": sample_rate}

src/free_claude_code/messaging/voice.py

@@ -2,6 +2,15 @@
 
 import asyncio
 from pathlib import Path
+from typing import Protocol
+
+
+class Transcriber(Protocol):
+    """Consumer-owned voice transcription boundary."""
+
+    async def transcribe(self, file_path: Path) -> str: ...
+
+    async def close(self) -> None: ...
 
 
 class PendingVoiceRegistry:
@@ -39,36 +48,3 @@ class PendingVoiceRegistry:
         async with self._lock:
             self._pending.pop((chat_id, voice_msg_id), None)
             self._pending.pop((chat_id, status_msg_id), None)
-
-
-class VoiceTranscriptionService:
-    """Run configured transcription backends off the event loop."""
-
-    def __init__(
-        self,
-        *,
-        huggingface_api_key: str = "",
-        nvidia_nim_api_key: str = "",
-    ) -> None:
-        self._huggingface_api_key = huggingface_api_key
-        self._nvidia_nim_api_key = nvidia_nim_api_key
-
-    async def transcribe(
-        self,
-        file_path: Path,
-        mime_type: str,
-        *,
-        whisper_model: str,
-        whisper_device: str,
-    ) -> str:
-        from .transcription import transcribe_audio
-
-        return await asyncio.to_thread(
-            transcribe_audio,
-            file_path,
-            mime_type,
-            whisper_model=whisper_model,
-            whisper_device=whisper_device,
-            huggingface_api_key=self._huggingface_api_key,
-            nvidia_nim_api_key=self._nvidia_nim_api_key,
-        )

src/free_claude_code/providers/cerebras/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import CEREBRAS_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -20,12 +21,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class CerebrasProvider(OpenAIChatTransport):
     """Cerebras API at ``https://api.cerebras.ai/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="CEREBRAS",
             base_url=config.base_url or CEREBRAS_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/cloudflare/client.py

@@ -17,6 +17,7 @@ from free_claude_code.providers.model_listing import (
     extract_openai_model_ids,
     model_infos_from_ids,
 )
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.http import maybe_await_aclose
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
@@ -59,7 +60,13 @@ def _cloudflare_account_api_url(api_root: str | None, account_id: str) -> str:
 class CloudflareProvider(OpenAIChatTransport):
     """Cloudflare Workers AI OpenAI-compatible chat provider."""
 
-    def __init__(self, config: ProviderConfig, *, account_id: str):
+    def __init__(
+        self,
+        config: ProviderConfig,
+        *,
+        account_id: str,
+        rate_limiter: ProviderRateLimiter,
+    ):
         base_url = cloudflare_ai_base_url(config.base_url, account_id)
         self._model_search_url = _cloudflare_model_search_url(
             config.base_url, account_id
@@ -78,6 +85,7 @@ class CloudflareProvider(OpenAIChatTransport):
             provider_name="CLOUDFLARE",
             base_url=base_url,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     async def cleanup(self) -> None:

src/free_claude_code/providers/codestral/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import CODESTRAL_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -20,12 +21,13 @@ class CodestralProvider(OpenAIChatTransport):
     Request shaping matches Mistral La Plateforme.
     """
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="CODESTRAL",
             base_url=config.base_url or CODESTRAL_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/cohere/client.py

@@ -7,6 +7,7 @@ from typing import Any
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import COHERE_DEFAULT_BASE
 from free_claude_code.providers.exceptions import InvalidRequestError
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -44,12 +45,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class CohereProvider(OpenAIChatTransport):
     """Cohere Compatibility API at ``https://api.cohere.ai/compatibility/v1``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="COHERE",
             base_url=config.base_url or COHERE_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/deepseek/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import DEEPSEEK_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import OpenAIChatTransport
 from free_claude_code.providers.transports.openai_chat.usage import usage_int
 
@@ -13,12 +14,13 @@ from .compat import build_deepseek_request_body
 class DeepSeekProvider(OpenAIChatTransport):
     """DeepSeek using ``https://api.deepseek.com`` Chat Completions."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="DEEPSEEK",
             base_url=config.base_url or DEEPSEEK_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/error_mapping.py

@@ -24,7 +24,7 @@ from free_claude_code.providers.exceptions import (
     ProviderError,
     RateLimitError,
 )
-from free_claude_code.providers.rate_limit import GlobalRateLimiter
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 
 _BODY_ATTR = "_fcc_provider_error_body"
 _BODY_TRUNCATED_ATTR = "_fcc_provider_error_body_truncated"
@@ -311,7 +311,7 @@ def map_stream_start_error(
     provider_name: str,
     read_timeout_s: float | None,
     request_id: str | None,
-    rate_limiter: GlobalRateLimiter | None = None,
+    rate_limiter: ProviderRateLimiter,
 ) -> ProviderError:
     """Map a final pre-start stream failure into an HTTP-serializable provider error.
 
@@ -339,17 +339,14 @@ def map_stream_start_error(
     return APIError(message, status_code=502, raw_error=str(error))
 
 
-def map_error(
-    e: Exception, *, rate_limiter: GlobalRateLimiter | None = None
-) -> Exception:
+def map_error(e: Exception, *, rate_limiter: ProviderRateLimiter) -> Exception:
     """Map OpenAI or HTTPX exception to specific ProviderError.
 
-    Streaming transports should pass their scoped limiter (``self._global_rate_limiter``)
-    so reactive 429 handling applies to the correct provider. Tests may omit
-    ``rate_limiter`` to use the process-wide singleton.
+    Streaming transports pass their owned limiter so reactive 429 handling
+    applies only to that provider instance.
     """
     message = get_user_facing_error_message(e)
-    limiter = rate_limiter or GlobalRateLimiter.get_instance()
+    limiter = rate_limiter
 
     if isinstance(e, openai.AuthenticationError):
         return AuthenticationError(message, raw_error=str(e))

src/free_claude_code/providers/fireworks/client.py

@@ -5,6 +5,7 @@ from typing import Any
 from free_claude_code.config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import FIREWORKS_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -27,12 +28,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class FireworksProvider(OpenAIChatTransport):
     """Fireworks AI using ``https://api.fireworks.ai/inference/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="FIREWORKS",
             base_url=config.base_url or FIREWORKS_BASE_URL,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/gemini/client.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import GEMINI_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -20,12 +21,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(provider_name="GEMINI")
 class GeminiProvider(OpenAIChatTransport):
     """Gemini API using ``https://generativelanguage.googleapis.com/v1beta/openai/``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="GEMINI",
             base_url=config.base_url or GEMINI_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
         self._tool_call_extra_content_by_id: dict[str, dict[str, Any]] = {}
 

src/free_claude_code/providers/github_models/client.py

@@ -12,6 +12,7 @@ from free_claude_code.providers.model_listing import (
     ProviderModelInfo,
     model_infos_from_ids,
 )
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.http import maybe_await_aclose
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
@@ -31,7 +32,7 @@ _REQUIRED_MODEL_CAPABILITIES = frozenset({"streaming", "tool-calling"})
 class GitHubModelsProvider(OpenAIChatTransport):
     """GitHub Models OpenAI-compatible inference provider."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         self._catalog_url = GITHUB_MODELS_CATALOG_URL
         self._model_list_client = httpx.AsyncClient(
             proxy=config.proxy or None,
@@ -47,6 +48,7 @@ class GitHubModelsProvider(OpenAIChatTransport):
             provider_name="GITHUB_MODELS",
             base_url=config.base_url or GITHUB_MODELS_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
             default_headers=_github_models_default_headers(),
         )
 

src/free_claude_code/providers/groq/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import GROQ_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -23,12 +24,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class GroqProvider(OpenAIChatTransport):
     """Groq API using ``https://api.groq.com/openai/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="GROQ",
             base_url=config.base_url or GROQ_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/huggingface/client.py

@@ -8,18 +8,20 @@ from free_claude_code.core.anthropic.conversion import OpenAIConversionError
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import HUGGINGFACE_DEFAULT_BASE
 from free_claude_code.providers.exceptions import InvalidRequestError
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import OpenAIChatTransport
 
 
 class HuggingFaceProvider(OpenAIChatTransport):
     """Hugging Face Inference Providers router at ``https://router.huggingface.co/v1``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="HUGGINGFACE",
             base_url=config.base_url or HUGGINGFACE_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/kimi/client.py

@@ -5,6 +5,7 @@ from typing import Any
 from free_claude_code.config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import KIMI_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -23,12 +24,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class KimiProvider(OpenAIChatTransport):
     """Kimi provider using ``https://api.moonshot.ai/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="KIMI",
             base_url=config.base_url or KIMI_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/llamacpp/client.py

@@ -2,6 +2,7 @@
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import LLAMACPP_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.anthropic_messages import (
     AnthropicMessagesTransport,
 )
@@ -10,9 +11,10 @@ from free_claude_code.providers.transports.anthropic_messages import (
 class LlamaCppProvider(AnthropicMessagesTransport):
     """Llama.cpp provider using native Anthropic Messages endpoint."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="LLAMACPP",
             default_base_url=LLAMACPP_DEFAULT_BASE,
+            rate_limiter=rate_limiter,
         )

src/free_claude_code/providers/lmstudio/client.py

@@ -26,6 +26,7 @@ from free_claude_code.core.anthropic.conversion import OpenAIConversionError
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import LMSTUDIO_DEFAULT_BASE
 from free_claude_code.providers.exceptions import InvalidRequestError
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import OpenAIChatTransport
 
 
@@ -39,12 +40,13 @@ class LMStudioProvider(OpenAIChatTransport):
     # dying mid-stream.
     _CONTEXT_CACHE_TTL_S = 30.0
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="LMSTUDIO",
             base_url=config.base_url or LMSTUDIO_DEFAULT_BASE,
             api_key=config.api_key or "lm-studio",
+            rate_limiter=rate_limiter,
         )
         self._loaded_context_cache: tuple[float, int | None] = (0.0, None)
 

src/free_claude_code/providers/minimax/client.py

@@ -5,6 +5,7 @@ from typing import Any
 from free_claude_code.config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import MINIMAX_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -21,12 +22,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class MiniMaxProvider(OpenAIChatTransport):
     """MiniMax using ``https://api.minimax.io/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="MINIMAX",
             base_url=config.base_url or MINIMAX_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/mistral/client.py

@@ -6,6 +6,7 @@ from loguru import logger
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import MISTRAL_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -25,12 +26,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(provider_name="MISTRAL")
 class MistralProvider(OpenAIChatTransport):
     """Mistral API using ``https://api.mistral.ai/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="MISTRAL",
             base_url=config.base_url or MISTRAL_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/nvidia_nim/client.py

@@ -9,6 +9,7 @@ from loguru import logger
 from free_claude_code.config.nim import NimSettings
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import NVIDIA_NIM_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import OpenAIChatTransport
 
 from .request_options import build_nim_request_body
@@ -26,12 +27,19 @@ from .tool_schema import (
 class NvidiaNimProvider(OpenAIChatTransport):
     """NVIDIA NIM provider using official OpenAI client."""
 
-    def __init__(self, config: ProviderConfig, *, nim_settings: NimSettings):
+    def __init__(
+        self,
+        config: ProviderConfig,
+        *,
+        nim_settings: NimSettings,
+        rate_limiter: ProviderRateLimiter,
+    ):
         super().__init__(
             config,
             provider_name="NIM",
             base_url=config.base_url or NVIDIA_NIM_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
         self._nim_settings = nim_settings
 

src/free_claude_code/providers/nvidia_nim/voice.py

@@ -1,5 +1,6 @@
 """NVIDIA NIM / Riva offline ASR for voice notes (provider-owned transport)."""
 
+import asyncio
 from pathlib import Path
 
 from loguru import logger
@@ -23,71 +24,90 @@ _NIM_ASR_MODEL_MAP: dict[str, tuple[str, str]] = {
 _RIVA_SERVER = "grpc.nvcf.nvidia.com:443"
 
 
-def transcribe_audio_file(
-    file_path: Path,
-    model: str,
-    *,
-    api_key: str,
-) -> str:
-    """Transcribe audio using NVIDIA NIM / Riva gRPC (offline recognition).
-
-    Args:
-        file_path: Path to encoded audio bytes readable by Riva.
-        model: Hugging Face-style NIM model id (see ``_NIM_ASR_MODEL_MAP``).
-        api_key: NVIDIA API key (Bearer token); must be non-empty.
-
-    Returns:
-        Transcript text, or ``(no speech detected)`` when empty.
-    """
-    key = (api_key or "").strip()
-    if not key:
-        raise ValueError(
-            "NVIDIA NIM transcription requires a non-empty nvidia_nim_api_key "
-            "(configure NVIDIA_NIM_API_KEY or pass api_key explicitly)."
-        )
-
-    try:
-        import riva.client
-    except ImportError as e:
-        raise ImportError(
-            "NVIDIA NIM transcription requires the voice extra. "
-            "Install with: uv sync --extra voice"
-        ) from e
-
-    model_config = _NIM_ASR_MODEL_MAP.get(model)
-    if not model_config:
-        raise ValueError(
-            f"No NVIDIA NIM config found for model: {model}. "
-            f"Supported models: {', '.join(_NIM_ASR_MODEL_MAP.keys())}"
+class NvidiaNimTranscriber:
+    """Own configured NVIDIA NIM / Riva transcription."""
+
+    def __init__(self, *, model: str, api_key: str) -> None:
+        self._model = model
+        self._key = api_key.strip()
+        self._lock = asyncio.Lock()
+        self._closed = False
+
+    async def transcribe(self, file_path: Path) -> str:
+        """Transcribe one audio file without blocking the event loop."""
+        async with self._lock:
+            if self._closed:
+                raise RuntimeError("NVIDIA NIM transcriber is closed.")
+            worker = asyncio.create_task(
+                asyncio.to_thread(self._transcribe_sync, file_path)
+            )
+            try:
+                return await asyncio.shield(worker)
+            except asyncio.CancelledError:
+                await _wait_for_thread_exit(worker)
+                raise
+
+    async def close(self) -> None:
+        """Close this stateless adapter to future work."""
+        self._closed = True
+        async with self._lock:
+            self._key = ""
+
+    def _transcribe_sync(self, file_path: Path) -> str:
+        if not self._key:
+            raise ValueError(
+                "NVIDIA NIM transcription requires a non-empty "
+                "nvidia_nim_api_key (configure NVIDIA_NIM_API_KEY)."
+            )
+        model_config = _NIM_ASR_MODEL_MAP.get(self._model)
+        if model_config is None:
+            raise ValueError(
+                f"No NVIDIA NIM config found for model: {self._model}. "
+                f"Supported models: {', '.join(_NIM_ASR_MODEL_MAP)}"
+            )
+        function_id, language_code = model_config
+        try:
+            import riva.client
+        except ImportError as exc:
+            raise ImportError(
+                "NVIDIA NIM transcription requires the voice extra. "
+                "Install with: uv sync --extra voice"
+            ) from exc
+
+        auth = riva.client.Auth(
+            use_ssl=True,
+            uri=_RIVA_SERVER,
+            metadata_args=[
+                ["function-id", function_id],
+                ["authorization", f"Bearer {self._key}"],
+            ],
         )
-    function_id, language_code = model_config
-
-    auth = riva.client.Auth(
-        use_ssl=True,
-        uri=_RIVA_SERVER,
-        metadata_args=[
-            ["function-id", function_id],
-            ["authorization", f"Bearer {key}"],
-        ],
-    )
-
-    asr_service = riva.client.ASRService(auth)
-
-    config = riva.client.RecognitionConfig(
-        language_code=language_code,
-        max_alternatives=1,
-        verbatim_transcripts=True,
-    )
-
-    with open(file_path, "rb") as f:
-        data = f.read()
-
-    response = asr_service.offline_recognize(data, config)
-
-    transcript = ""
-    results = getattr(response, "results", None)
-    if results and results[0].alternatives:
-        transcript = results[0].alternatives[0].transcript
-
-    logger.debug(f"NIM transcription: {len(transcript)} chars")
-    return transcript or "(no speech detected)"
+        try:
+            asr_service = riva.client.ASRService(auth)
+            config = riva.client.RecognitionConfig(
+                language_code=language_code,
+                max_alternatives=1,
+                verbatim_transcripts=True,
+            )
+            data = file_path.read_bytes()
+            response = asr_service.offline_recognize(data, config)
+
+            transcript = ""
+            results = getattr(response, "results", None)
+            if results and results[0].alternatives:
+                transcript = results[0].alternatives[0].transcript
+            logger.debug("NIM transcription: {} chars", len(transcript))
+            return transcript or "(no speech detected)"
+        finally:
+            auth.channel.close()
+
+
+async def _wait_for_thread_exit(worker: asyncio.Task[str]) -> None:
+    """Wait through repeated caller cancellation without cancelling thread work."""
+    while not worker.done():
+        try:
+            await asyncio.shield(asyncio.wait((worker,)))
+        except asyncio.CancelledError:
+            continue
+    if not worker.cancelled():
+        worker.exception()

src/free_claude_code/providers/ollama/client.py

@@ -5,6 +5,7 @@ import httpx
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import OLLAMA_DEFAULT_BASE
 from free_claude_code.providers.model_listing import extract_ollama_model_ids
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.anthropic_messages import (
     AnthropicMessagesTransport,
 )
@@ -13,11 +14,12 @@ from free_claude_code.providers.transports.anthropic_messages import (
 class OllamaProvider(AnthropicMessagesTransport):
     """Ollama provider using native Anthropic Messages API."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="OLLAMA",
             default_base_url=OLLAMA_DEFAULT_BASE,
+            rate_limiter=rate_limiter,
         )
         self._api_key = config.api_key or "ollama"
 

src/free_claude_code/providers/open_router/client.py

@@ -13,6 +13,7 @@ from free_claude_code.providers.model_listing import (
     extract_openrouter_tool_model_ids,
     extract_openrouter_tool_model_infos,
 )
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -33,12 +34,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class OpenRouterProvider(OpenAIChatTransport):
     """OpenRouter provider using the OpenAI-compatible Chat Completions API."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="OPENROUTER",
             base_url=config.base_url or OPENROUTER_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/opencode/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import OPENCODE_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -14,12 +15,19 @@ from free_claude_code.providers.transports.openai_chat import (
 class OpenCodeProvider(OpenAIChatTransport):
     """OpenCode Zen provider using ``https://opencode.ai/zen/v1/chat/completions``."""
 
-    def __init__(self, config: ProviderConfig, provider_name: str = "OPENCODE"):
+    def __init__(
+        self,
+        config: ProviderConfig,
+        provider_name: str = "OPENCODE",
+        *,
+        rate_limiter: ProviderRateLimiter,
+    ):
         super().__init__(
             config,
             provider_name=provider_name,
             base_url=config.base_url or OPENCODE_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
         self._request_policy = OpenAIChatRequestPolicy(provider_name=provider_name)
 

src/free_claude_code/providers/rate_limit.py

@@ -1,11 +1,11 @@
-"""Global rate limiter for API requests."""
+"""Provider-owned upstream rate limiting and retry policy."""
 
 import asyncio
 import random
 import time
 from collections.abc import AsyncIterator, Callable
 from contextlib import asynccontextmanager
-from typing import Any, ClassVar, TypeVar
+from typing import Any, TypeVar
 
 import httpx
 import openai
@@ -58,11 +58,13 @@ def retryable_upstream_transport_error(exc: BaseException) -> bool:
     )
 
 
-class GlobalRateLimiter:
+class ProviderRateLimiter:
     """
-    Global singleton rate limiter that blocks all requests
-    when a rate limit error is encountered (reactive) and
-    throttles requests (proactive) using a strict rolling window.
+    Rate limiter owned by one provider instance.
+
+    Blocks that provider's requests when a rate-limit error is encountered
+    (reactive) and throttles its requests with a strict rolling window
+    (proactive).
 
     Optionally enforces a max_concurrency cap: at most N provider streams
     may be open simultaneously, independent of the sliding window.
@@ -72,19 +74,12 @@ class GlobalRateLimiter:
     Concurrency limit - caps simultaneously open streams.
     """
 
-    _instance: ClassVar[GlobalRateLimiter | None] = None
-    _scoped_instances: ClassVar[dict[str, GlobalRateLimiter]] = {}
-
     def __init__(
         self,
         rate_limit: int = 40,
         rate_window: float = 60.0,
         max_concurrency: int = 5,
     ):
-        # Prevent re-initialization on singleton reuse
-        if hasattr(self, "_initialized"):
-            return
-
         if rate_limit <= 0:
             raise ValueError("rate_limit must be > 0")
         if rate_window <= 0:
@@ -100,69 +95,10 @@ class GlobalRateLimiter:
         )
         self._blocked_until: float = 0
         self._concurrency_sem = asyncio.Semaphore(max_concurrency)
-        self._initialized = True
-
         logger.info(
-            f"GlobalRateLimiter (Provider) initialized ({rate_limit} req / {rate_window}s, max_concurrency={max_concurrency})"
-        )
-
-    @classmethod
-    def get_instance(
-        cls,
-        rate_limit: int | None = None,
-        rate_window: float | None = None,
-        max_concurrency: int = 5,
-    ) -> GlobalRateLimiter:
-        """Get or create the singleton instance.
-
-        Args:
-            rate_limit: Requests per window (only used on first creation)
-            rate_window: Window in seconds (only used on first creation)
-            max_concurrency: Max simultaneous open streams (only used on first creation)
-        """
-        if cls._instance is None:
-            cls._instance = cls(
-                rate_limit=rate_limit or 40,
-                rate_window=rate_window or 60.0,
-                max_concurrency=max_concurrency,
-            )
-        return cls._instance
-
-    @classmethod
-    def get_scoped_instance(
-        cls,
-        scope: str,
-        *,
-        rate_limit: int | None = None,
-        rate_window: float | None = None,
-        max_concurrency: int = 5,
-    ) -> GlobalRateLimiter:
-        """Get or create a provider-scoped limiter instance."""
-        if not scope:
-            raise ValueError("scope must be non-empty")
-        desired_rate_limit = rate_limit or 40
-        desired_rate_window = float(rate_window or 60.0)
-        existing = cls._scoped_instances.get(scope)
-        if existing and existing.matches_config(
-            desired_rate_limit, desired_rate_window, max_concurrency
-        ):
-            return existing
-        if existing:
-            logger.info(
-                "Rebuilding provider rate limiter for updated scope '{}'", scope
-            )
-        cls._scoped_instances[scope] = cls(
-            rate_limit=desired_rate_limit,
-            rate_window=desired_rate_window,
-            max_concurrency=max_concurrency,
+            "ProviderRateLimiter initialized "
+            f"({rate_limit} req / {rate_window}s, max_concurrency={max_concurrency})"
         )
-        return cls._scoped_instances[scope]
-
-    @classmethod
-    def reset_instance(cls) -> None:
-        """Reset singleton (for testing)."""
-        cls._instance = None
-        cls._scoped_instances = {}
 
     async def wait_if_blocked(self) -> bool:
         """
@@ -177,7 +113,7 @@ class GlobalRateLimiter:
         if now < self._blocked_until:
             wait_time = self._blocked_until - now
             logger.warning(
-                f"Global provider rate limit active (reactive), waiting {wait_time:.1f}s..."
+                f"Provider rate limit active (reactive), waiting {wait_time:.1f}s..."
             )
             await asyncio.sleep(wait_time)
             waited_reactively = True
@@ -197,28 +133,18 @@ class GlobalRateLimiter:
 
     def set_blocked(self, seconds: float = 60) -> None:
         """
-        Set global block for specified seconds (reactive).
+        Set this provider's block for the specified seconds (reactive).
 
         Args:
             seconds: How long to block (default 60s)
         """
         self._blocked_until = time.monotonic() + seconds
-        logger.warning(f"Global provider rate limit set for {seconds:.1f}s (reactive)")
+        logger.warning(f"Provider rate limit set for {seconds:.1f}s (reactive)")
 
     def is_blocked(self) -> bool:
         """Check if currently reactively blocked."""
         return time.monotonic() < self._blocked_until
 
-    def matches_config(
-        self, rate_limit: int, rate_window: float, max_concurrency: int
-    ) -> bool:
-        """Return whether this limiter matches the requested runtime config."""
-        return (
-            self._rate_limit == rate_limit
-            and self._rate_window == float(rate_window)
-            and self._max_concurrency == max_concurrency
-        )
-
     def remaining_wait(self) -> float:
         """Get remaining reactive wait time in seconds."""
         return max(0.0, self._blocked_until - time.monotonic())

src/free_claude_code/providers/runtime/cache.py

@@ -1,5 +1,6 @@
 """Provider instance cache and cleanup."""
 
+import asyncio
 from collections.abc import Callable, MutableMapping
 
 from free_claude_code.config.settings import Settings
@@ -35,17 +36,18 @@ class ProviderCache:
         return self._providers[provider_id]
 
     async def cleanup(self) -> None:
-        """Clean up every cached provider, then clear the cache."""
+        """Clean every cached provider, retaining unfinished entries for retry."""
         items = list(self._providers.items())
         errors: list[Exception] = []
-        try:
-            for _provider_id, provider in items:
-                try:
-                    await provider.cleanup()
-                except Exception as exc:
-                    errors.append(exc)
-        finally:
-            self._providers.clear()
+        for provider_id, provider in items:
+            try:
+                await provider.cleanup()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                errors.append(exc)
+            else:
+                self._providers.pop(provider_id, None)
         if len(errors) == 1:
             raise errors[0]
         if len(errors) > 1:

src/free_claude_code/providers/runtime/factory.py

@@ -9,156 +9,265 @@ from free_claude_code.config.provider_catalog import (
 from free_claude_code.config.settings import Settings
 from free_claude_code.providers.base import BaseProvider, ProviderConfig
 from free_claude_code.providers.exceptions import UnknownProviderTypeError
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 
 from .config import build_provider_config
 
-ProviderFactory = Callable[[ProviderConfig, Settings], BaseProvider]
+ProviderFactory = Callable[
+    [ProviderConfig, Settings, ProviderRateLimiter], BaseProvider
+]
 
 
-def _create_nvidia_nim(config: ProviderConfig, settings: Settings) -> BaseProvider:
+def _create_nvidia_nim(
+    config: ProviderConfig,
+    settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.nvidia_nim import NvidiaNimProvider
 
-    return NvidiaNimProvider(config, nim_settings=settings.nim)
+    return NvidiaNimProvider(
+        config,
+        nim_settings=settings.nim,
+        rate_limiter=rate_limiter,
+    )
 
 
-def _create_open_router(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_open_router(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.open_router import OpenRouterProvider
 
-    return OpenRouterProvider(config)
+    return OpenRouterProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_mistral(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_mistral(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.mistral import MistralProvider
 
-    return MistralProvider(config)
+    return MistralProvider(config, rate_limiter=rate_limiter)
 
 
 def _create_mistral_codestral(
-    config: ProviderConfig, _settings: Settings
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
 ) -> BaseProvider:
     from free_claude_code.providers.codestral import CodestralProvider
 
-    return CodestralProvider(config)
+    return CodestralProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_deepseek(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_deepseek(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.deepseek import DeepSeekProvider
 
-    return DeepSeekProvider(config)
+    return DeepSeekProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_lmstudio(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_lmstudio(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.lmstudio import LMStudioProvider
 
-    return LMStudioProvider(config)
+    return LMStudioProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_llamacpp(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_llamacpp(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.llamacpp import LlamaCppProvider
 
-    return LlamaCppProvider(config)
+    return LlamaCppProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_ollama(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_ollama(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.ollama import OllamaProvider
 
-    return OllamaProvider(config)
+    return OllamaProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_kimi(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_kimi(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.kimi import KimiProvider
 
-    return KimiProvider(config)
+    return KimiProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_wafer(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_wafer(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.wafer import WaferProvider
 
-    return WaferProvider(config)
+    return WaferProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_minimax(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_minimax(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.minimax import MiniMaxProvider
 
-    return MiniMaxProvider(config)
+    return MiniMaxProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_opencode(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_opencode(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.opencode import OpenCodeProvider
 
-    return OpenCodeProvider(config)
+    return OpenCodeProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_opencode_go(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_opencode_go(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.opencode import OpenCodeProvider
 
-    return OpenCodeProvider(config, provider_name="OPENCODE_GO")
+    return OpenCodeProvider(
+        config,
+        provider_name="OPENCODE_GO",
+        rate_limiter=rate_limiter,
+    )
 
 
-def _create_vercel(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_vercel(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.vercel import VercelProvider
 
-    return VercelProvider(config)
+    return VercelProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_huggingface(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_huggingface(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.huggingface import HuggingFaceProvider
 
-    return HuggingFaceProvider(config)
+    return HuggingFaceProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_cohere(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_cohere(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.cohere import CohereProvider
 
-    return CohereProvider(config)
+    return CohereProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_github_models(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_github_models(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.github_models import GitHubModelsProvider
 
-    return GitHubModelsProvider(config)
+    return GitHubModelsProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_zai(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_zai(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.zai import ZaiProvider
 
-    return ZaiProvider(config)
+    return ZaiProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_fireworks(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_fireworks(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.fireworks import FireworksProvider
 
-    return FireworksProvider(config)
+    return FireworksProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_cloudflare(config: ProviderConfig, settings: Settings) -> BaseProvider:
+def _create_cloudflare(
+    config: ProviderConfig,
+    settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.cloudflare import CloudflareProvider
 
-    return CloudflareProvider(config, account_id=settings.cloudflare_account_id)
+    return CloudflareProvider(
+        config,
+        account_id=settings.cloudflare_account_id,
+        rate_limiter=rate_limiter,
+    )
 
 
-def _create_gemini(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_gemini(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.gemini import GeminiProvider
 
-    return GeminiProvider(config)
+    return GeminiProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_groq(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_groq(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.groq import GroqProvider
 
-    return GroqProvider(config)
+    return GroqProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_sambanova(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_sambanova(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.sambanova import SambaNovaProvider
 
-    return SambaNovaProvider(config)
+    return SambaNovaProvider(config, rate_limiter=rate_limiter)
 
 
-def _create_cerebras(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+def _create_cerebras(
+    config: ProviderConfig,
+    _settings: Settings,
+    rate_limiter: ProviderRateLimiter,
+) -> BaseProvider:
     from free_claude_code.providers.cerebras import CerebrasProvider
 
-    return CerebrasProvider(config)
+    return CerebrasProvider(config, rate_limiter=rate_limiter)
 
 
 PROVIDER_FACTORIES: dict[str, ProviderFactory] = {
@@ -210,4 +319,10 @@ def create_provider(provider_id: str, settings: Settings) -> BaseProvider:
     factory = PROVIDER_FACTORIES.get(provider_id)
     if factory is None:
         raise AssertionError(f"Unhandled provider descriptor: {provider_id}")
-    return factory(build_provider_config(descriptor, settings), settings)
+    config = build_provider_config(descriptor, settings)
+    rate_limiter = ProviderRateLimiter(
+        rate_limit=config.rate_limit or 40,
+        rate_window=config.rate_window or 60.0,
+        max_concurrency=config.max_concurrency,
+    )
+    return factory(config, settings, rate_limiter)

src/free_claude_code/providers/sambanova/client.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from free_claude_code.providers.base import ProviderConfig
 from free_claude_code.providers.defaults import SAMBANOVA_DEFAULT_BASE
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.openai_chat import (
     OpenAIChatRequestPolicy,
     OpenAIChatTransport,
@@ -19,12 +20,13 @@ _REQUEST_POLICY = OpenAIChatRequestPolicy(
 class SambaNovaProvider(OpenAIChatTransport):
     """SambaNova Cloud API at ``https://api.sambanova.ai/v1``."""
 
-    def __init__(self, config: ProviderConfig):
+    def __init__(self, config: ProviderConfig, *, rate_limiter: ProviderRateLimiter):
         super().__init__(
             config,
             provider_name="SAMBANOVA",
             base_url=config.base_url or SAMBANOVA_DEFAULT_BASE,
             api_key=config.api_key,
+            rate_limiter=rate_limiter,
         )
 
     def _build_request_body(

src/free_claude_code/providers/transports/anthropic_messages/recovery.py

@@ -48,10 +48,8 @@ class AnthropicMessagesRecovery:
         for attempt in range(MIDSTREAM_RECOVERY_ATTEMPTS):
             response: httpx.Response | None = None
             try:
-                response = (
-                    await self._transport._global_rate_limiter.execute_with_retry(
-                        self._transport._validated_stream_send, body, req_tag=req_tag
-                    )
+                response = await self._transport._rate_limiter.execute_with_retry(
+                    self._transport._validated_stream_send, body, req_tag=req_tag
                 )
                 state = self._transport._new_stream_state(
                     None, thinking_enabled=thinking_enabled

src/free_claude_code/providers/transports/anthropic_messages/stream.py

@@ -91,16 +91,14 @@ class AnthropicMessagesStreamAdapter:
         ledger = self._new_ledger()
         recovery = RecoveryController(provider_name=tag, request_id=self._request_id)
 
-        async with self._transport._global_rate_limiter.concurrency_slot():
+        async with self._transport._rate_limiter.concurrency_slot():
             while True:
                 stream_opened = False
                 try:
-                    response = (
-                        await self._transport._global_rate_limiter.execute_with_retry(
-                            self._transport._validated_stream_send,
-                            body,
-                            req_tag=req_tag,
-                        )
+                    response = await self._transport._rate_limiter.execute_with_retry(
+                        self._transport._validated_stream_send,
+                        body,
+                        req_tag=req_tag,
                     )
                     stream_opened = True
                     chunk_count = 0
@@ -261,7 +259,7 @@ class AnthropicMessagesStreamAdapter:
                             provider_name=tag,
                             read_timeout_s=self._transport._config.http_read_timeout,
                             request_id=self._request_id,
-                            rate_limiter=self._transport._global_rate_limiter,
+                            rate_limiter=self._transport._rate_limiter,
                         ) from error
                     return
                 finally:

src/free_claude_code/providers/transports/anthropic_messages/transport.py

@@ -20,7 +20,7 @@ from free_claude_code.providers.model_listing import (
     extract_openai_model_ids,
     model_infos_from_ids,
 )
-from free_claude_code.providers.rate_limit import GlobalRateLimiter
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 from free_claude_code.providers.transports.http import maybe_await_aclose
 
 from .http import model_list_json, raise_for_status_with_body
@@ -44,18 +44,14 @@ class AnthropicMessagesTransport(BaseProvider):
         *,
         provider_name: str,
         default_base_url: str,
+        rate_limiter: ProviderRateLimiter,
     ):
         super().__init__(config)
         self._provider_name = provider_name
         self._api_key = config.api_key
         self._base_url = (config.base_url or default_base_url).rstrip("/")
         self._request_policy = NativeMessagesRequestPolicy(provider_name=provider_name)
-        self._global_rate_limiter = GlobalRateLimiter.get_scoped_instance(
-            provider_name.lower(),
-            rate_limit=config.rate_limit,
-            rate_window=config.rate_window,
-            max_concurrency=config.max_concurrency,
-        )
+        self._rate_limiter = rate_limiter
         self._client = httpx.AsyncClient(
             base_url=self._base_url,
             proxy=config.proxy or None,
@@ -182,7 +178,7 @@ class AnthropicMessagesTransport(BaseProvider):
         self, error: Exception, request_id: str | None
     ) -> tuple[Exception, str]:
         """Map an exception into a user-facing provider error message."""
-        mapped_error = map_error(error, rate_limiter=self._global_rate_limiter)
+        mapped_error = map_error(error, rate_limiter=self._rate_limiter)
         return (
             mapped_error,
             user_visible_message_for_mapped_provider_error(

src/free_claude_code/providers/transports/openai_chat/stream.py

@@ -104,7 +104,7 @@ class OpenAIChatStreamAdapter:
         tool_argument_aliases: dict[str, dict[str, str]] = {}
         tool_argument_alias_buffers: dict[int, str] = {}
 
-        async with self._transport._global_rate_limiter.concurrency_slot():
+        async with self._transport._rate_limiter.concurrency_slot():
             while True:
                 if not ledger.message_started:
                     for event in hold_event(ledger.message_start()):
@@ -306,7 +306,7 @@ class OpenAIChatStreamAdapter:
                             provider_name=tag,
                             read_timeout_s=self._transport._config.http_read_timeout,
                             request_id=self._request_id,
-                            rate_limiter=self._transport._global_rate_limiter,
+                            rate_limiter=self._transport._rate_limiter,
                         ) from error
                     for event in ledger.terminal_error_tail(
                         error_message,

src/free_claude_code/providers/transports/openai_chat/transport.py

@@ -16,7 +16,7 @@ from free_claude_code.providers.error_mapping import (
     user_visible_message_for_mapped_provider_error,
 )
 from free_claude_code.providers.model_listing import extract_openai_model_ids
-from free_claude_code.providers.rate_limit import GlobalRateLimiter
+from free_claude_code.providers.rate_limit import ProviderRateLimiter
 
 from .output_cap import clamp_output_tokens, parse_output_token_cap
 from .stream import OpenAIChatStreamAdapter
@@ -33,6 +33,7 @@ class OpenAIChatTransport(BaseProvider):
         provider_name: str,
         base_url: str,
         api_key: str,
+        rate_limiter: ProviderRateLimiter,
         default_headers: Mapping[str, str] | None = None,
     ):
         super().__init__(config)
@@ -42,12 +43,7 @@ class OpenAIChatTransport(BaseProvider):
         # Learned per-model output-token caps from upstream 400 rejections, so
         # later requests clamp proactively instead of paying the 400 each time.
         self._model_output_caps: dict[str, int] = {}
-        self._global_rate_limiter = GlobalRateLimiter.get_scoped_instance(
-            provider_name.lower(),
-            rate_limit=config.rate_limit,
-            rate_window=config.rate_window,
-            max_concurrency=config.max_concurrency,
-        )
+        self._rate_limiter = rate_limiter
         http_client = None
         if config.proxy:
             http_client = httpx.AsyncClient(
@@ -125,7 +121,7 @@ class OpenAIChatTransport(BaseProvider):
         while True:
             try:
                 create_body = self._prepare_create_body(body)
-                stream = await self._global_rate_limiter.execute_with_retry(
+                stream = await self._rate_limiter.execute_with_retry(
                     self._client.chat.completions.create, **create_body, stream=True
                 )
                 return stream, body
@@ -198,7 +194,7 @@ class OpenAIChatTransport(BaseProvider):
     def _map_error_details(
         self, error: Exception, request_id: str | None
     ) -> tuple[Exception, str]:
-        mapped_error = map_error(error, rate_limiter=self._global_rate_limiter)
+        mapped_error = map_error(error, rate_limiter=self._rate_limiter)
         return (
             mapped_error,
             user_visible_message_for_mapped_provider_error(