Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions e2e/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,6 @@ def __init__(
ProviderConfig(
api_key="browser-test",
base_url="https://provider.invalid/v1",
rate_limit=1_000,
rate_window=1,
max_concurrency=100,
http_read_timeout=1.0,
http_write_timeout=1.0,
http_connect_timeout=1.0,
Expand Down Expand Up @@ -252,15 +249,20 @@ def admin_base_url(
),
}

async def fixture_provider(provider_id: str, _settings: Settings) -> BaseProvider:
async def fixture_provider(
provider_id: str, _settings: Settings, _admission_registry
) -> BaseProvider:
if provider_id not in providers:
raise AssertionError(f"Missing browser fixture provider: {provider_id}")
return providers[provider_id]

manager = ProviderRuntimeManager(
get_settings(),
runtime_factory=lambda snapshot: ProviderRuntime(
snapshot, dict(providers), provider_constructor=fixture_provider
runtime_factory=lambda snapshot, admission_registry: ProviderRuntime(
snapshot,
admission_registry,
dict(providers),
provider_constructor=fixture_provider,
),
)
runtime = ApplicationRuntime(
Expand Down
2 changes: 1 addition & 1 deletion scripts/install.ps1
Original file line number Diff line number Diff line change
Expand Up @@ -1006,7 +1006,7 @@ function Install-Hermes {
Invoke-DownloadedPowerShellInstaller `
-Url $HermesInstallUrl `
-Name "Hermes Agent" `
-ScriptArguments @("-NonInteractive", "-SkipSetup")
-ScriptArguments @("-NonInteractive")
Add-KnownBinDirectories
}

Expand Down
22 changes: 16 additions & 6 deletions smoke/installers/test_installers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2327,15 +2327,15 @@ def powershell_harness(
encoding="utf-8",
)
(fixtures / "hermes-installer.ps1").write_text(
r"""param(
[switch] $NonInteractive,
[switch] $SkipSetup
r"""[CmdletBinding(PositionalBinding=$false)]
param(
[switch] $NonInteractive
)
if ($env:FAIL_STEP -eq "hermes-install") { exit 65 }
$bin = Join-Path $env:LOCALAPPDATA "hermes\hermes-agent\bin"
New-Item -ItemType Directory -Force -Path $bin | Out-Null
Copy-Item (Join-Path $env:FAKE_FIXTURES "hermes-command.cmd") (Join-Path $bin "hermes.cmd") -Force
Add-Content -LiteralPath $env:CALL_LOG -Value "hermes-install:${NonInteractive}:${SkipSetup}"
Add-Content -LiteralPath $env:CALL_LOG -Value "hermes-install:${NonInteractive}"
""",
encoding="utf-8",
)
Expand Down Expand Up @@ -3173,7 +3173,7 @@ def test_install_ps1_fresh_install_is_verified(
)
assert calls.index("npm:install -g cline") < calls.index("cline:--version")
assert any("hermes-agent.nousresearch.com/install.ps1" in call for call in calls)
assert "hermes-install:True:True" in calls
assert "hermes-install:True" in calls
assert calls.index("npm:install -g @deepseek-ai/dsh@0.1.0-rc.8") < calls.index(
"dsh:--version"
)
Expand Down Expand Up @@ -3251,7 +3251,7 @@ def test_install_ps1_discovers_grok_in_custom_bin_directory(
("client", "install_call"),
[
("cline", "npm:install -g cline"),
("hermes", "hermes-install:True:True"),
("hermes", "hermes-install:True"),
("grok", "grok-install"),
(
"aider",
Expand Down Expand Up @@ -3538,6 +3538,16 @@ def test_install_ps1_stops_when_selected_dsh_install_fails(
_assert_uv_ready_without_fcc_install(powershell_harness.calls())


def test_install_ps1_installs_hermes_noninteractively(
powershell_harness: PowerShellHarness,
) -> None:
result = powershell_harness.run_functions("Ensure-Hermes")

assert result.returncode == 0, result.stderr
calls = powershell_harness.calls()
assert calls.index("hermes-install:True") < calls.index("hermes:--version")


def test_install_ps1_rejects_unsupported_hermes_architecture_before_download(
powershell_harness: PowerShellHarness,
) -> None:
Expand Down
8 changes: 5 additions & 3 deletions src/free_claude_code/config/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -584,12 +584,14 @@ def validate_provider_references(self) -> Settings:
default=None, validation_alias="OLLAMA_CLOUD_PROXY"
)
# ==================== Provider Rate Limiting ====================
provider_rate_limit: int = Field(default=1, validation_alias="PROVIDER_RATE_LIMIT")
provider_rate_limit: int = Field(
default=1, gt=0, validation_alias="PROVIDER_RATE_LIMIT"
)
provider_rate_window: int = Field(
default=2, validation_alias="PROVIDER_RATE_WINDOW"
default=2, gt=0, validation_alias="PROVIDER_RATE_WINDOW"
)
provider_max_concurrency: int = Field(
default=2, validation_alias="PROVIDER_MAX_CONCURRENCY"
default=2, gt=0, validation_alias="PROVIDER_MAX_CONCURRENCY"
)
provider_progress_timeout: float = Field(
default=600.0,
Expand Down
103 changes: 73 additions & 30 deletions src/free_claude_code/providers/admission.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import random
import time
import uuid
from collections import deque
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
Expand All @@ -14,8 +15,8 @@

from loguru import logger

from free_claude_code.core.rate_limit import StrictSlidingWindowLimiter
from free_claude_code.core.trace import trace_event
from free_claude_code.providers.admission_policy import ProviderAdmissionLimits
from free_claude_code.providers.failure_policy import (
ProviderFailureOverride,
ProviderRecoveryExhausted,
Expand Down Expand Up @@ -416,12 +417,7 @@ def __init__(
max_delay: float = DEFAULT_UPSTREAM_MAX_DELAY,
jitter: float = DEFAULT_UPSTREAM_JITTER,
) -> None:
if rate_limit <= 0:
raise ValueError("rate_limit must be > 0")
if rate_window <= 0:
raise ValueError("rate_window must be > 0")
if max_concurrency <= 0:
raise ValueError("max_concurrency must be > 0")
limits = ProviderAdmissionLimits(rate_limit, rate_window, max_concurrency)
if max_attempts <= 0:
raise ValueError("max_attempts must be > 0")
if base_delay < 0:
Expand All @@ -436,10 +432,11 @@ def __init__(
self._base_delay = base_delay
self._max_delay = max_delay
self._jitter = jitter
self._proactive_limiter = StrictSlidingWindowLimiter(
rate_limit, float(rate_window)
)
self._concurrency_sem = asyncio.Semaphore(max_concurrency)
self._limits = limits
self._admission_times: deque[float] = deque()
self._last_discarded_at: float | None = None
self._active_attempts = 0
self._capacity_changed = asyncio.Event()
self._condition = asyncio.Condition()
self._episode: _RecoveryEpisode | None = None
self._next_generation = 1
Expand Down Expand Up @@ -485,19 +482,11 @@ async def _open_attempt(
slot_acquired = False
claim: _AttemptClaim | None = None
try:
admitted = await self._proactive_limiter.acquire_if(
lambda permit=permit: self._permit_is_current(execution, permit)
)
admitted = await self._acquire_capacity(execution, permit)
if not admitted:
await self._abandon_probe_permit(execution, permit)
continue
await self._concurrency_sem.acquire()
slot_acquired = True
if not self._permit_is_current(execution, permit):
self._concurrency_sem.release()
slot_acquired = False
await self._abandon_probe_permit(execution, permit)
continue
claim = execution._claim_attempt(operation_kind)
attempt = ProviderAttempt(self, execution, permit, claim)
trace_event(
Expand All @@ -518,7 +507,7 @@ async def _open_attempt(
if claim is not None:
execution._close_attempt(claim)
if slot_acquired:
self._concurrency_sem.release()
self._release_concurrency()
await self._abandon_probe_permit(execution, permit)
raise

Expand Down Expand Up @@ -639,7 +628,7 @@ async def _attempt_accepted(
if episode is None:
return
self._episode = None
self._condition.notify_all()
self._notify_recovery_waiters()
trace_event(
stage="provider",
event="provider.recovery.closed",
Expand All @@ -664,7 +653,7 @@ async def _attempt_corrected(
return
episode.probe_active = False
episode.ready_at = time.monotonic()
self._condition.notify_all()
self._notify_recovery_waiters()

async def _attempt_rejected(
self,
Expand All @@ -679,7 +668,7 @@ async def _attempt_rejected(
if episode is None:
return
self._episode = None
self._condition.notify_all()
self._notify_recovery_waiters()
trace_event(
stage="provider",
event="provider.recovery.closed",
Expand All @@ -705,7 +694,7 @@ async def _attempt_abandoned(
episode.leader = None
episode.probe_active = False
episode.ready_at = time.monotonic()
self._condition.notify_all()
self._notify_recovery_waiters()

async def _attempt_failed(
self,
Expand Down Expand Up @@ -740,7 +729,7 @@ async def _attempt_failed(
execution._fail_recovery(episode.last_error)
else:
episode.waiters.add(execution)
self._condition.notify_all()
self._notify_recovery_waiters()
elif episode is None or matching_probe is not None:
terminal_delay = self._retry_delay(
error,
Expand All @@ -761,7 +750,7 @@ async def _attempt_failed(
waiter._fail_recovery(error)
episode.waiters.clear()
exhausted_episode = True
self._condition.notify_all()
self._notify_recovery_waiters()

label = self._failure_label(status, error)
if became_leader:
Expand Down Expand Up @@ -871,7 +860,7 @@ async def _abandon_probe_permit(
episode.leader = None
episode.probe_active = False
episode.ready_at = time.monotonic()
self._condition.notify_all()
self._notify_recovery_waiters()

async def _abandon_waiting_leader(
self,
Expand All @@ -888,10 +877,64 @@ async def _abandon_waiting_leader(
):
return
episode.leader = None
self._condition.notify_all()
self._notify_recovery_waiters()

def reconfigure(self, limits: ProviderAdmissionLimits) -> None:
"""Update capacity without resetting occupancy, history, or recovery."""
if limits != self._limits:
self._limits = limits
self._wake_capacity()

def _wake_capacity(self) -> None:
changed = self._capacity_changed
self._capacity_changed = asyncio.Event()
changed.set()

def _notify_recovery_waiters(self) -> None:
self._condition.notify_all()
self._wake_capacity()

async def _acquire_capacity(
self, execution: ProviderExecution, permit: _GatePermit
) -> bool:
while self._permit_is_current(execution, permit):
changed = self._capacity_changed
now = time.monotonic()
limits = self._limits
cutoff = now - limits.rate_window
while self._admission_times and self._admission_times[0] <= cutoff:
self._last_discarded_at = self._admission_times.popleft()

ready_at = now
if self._last_discarded_at is not None:
ready_at = max(ready_at, self._last_discarded_at + limits.rate_window)
if len(self._admission_times) >= limits.rate_limit:
ready_at = max(
ready_at,
self._admission_times[-limits.rate_limit] + limits.rate_window,
)
if self._active_attempts < limits.max_concurrency and ready_at <= now:
self._admission_times.append(now)
self._active_attempts += 1
return True

# Occupancy changes explicitly wake us. Avoid an expired rate timer
# spinning while all concurrency slots remain occupied.
delay = (
max(0.0, ready_at - now)
if self._active_attempts < limits.max_concurrency
else None
)
try:
async with asyncio.timeout(delay):
await changed.wait()
except TimeoutError:
pass
return False

def _release_concurrency(self) -> None:
self._concurrency_sem.release()
self._active_attempts -= 1
self._wake_capacity()

def _retry_delay(self, error: Exception, attempt: int) -> float:
exponent = max(0, attempt - 1)
Expand Down
31 changes: 31 additions & 0 deletions src/free_claude_code/providers/admission_policy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
"""Validated capacity policy, independent of clients and controller lifetime."""

import math
from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from free_claude_code.config.settings import Settings


@dataclass(frozen=True, slots=True)
class ProviderAdmissionLimits:
rate_limit: int
rate_window: float
max_concurrency: int

def __post_init__(self) -> None:
for name in ("rate_limit", "max_concurrency"):
value = getattr(self, name)
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise ValueError(f"{name} must be a positive integer")
if not math.isfinite(self.rate_window) or self.rate_window <= 0:
raise ValueError("rate_window must be finite and > 0")

@classmethod
def from_settings(cls, settings: Settings) -> ProviderAdmissionLimits:
return cls(
settings.provider_rate_limit,
settings.provider_rate_window,
settings.provider_max_concurrency,
)
Loading
Loading