diff --git a/src/operations_center/execution/coordinator.py b/src/operations_center/execution/coordinator.py index 44f82a70..38d9bafe 100644 --- a/src/operations_center/execution/coordinator.py +++ b/src/operations_center/execution/coordinator.py @@ -25,6 +25,11 @@ from datetime import UTC, datetime +from operations_center.backends._capacity_classifier import classify_capacity_exhaustion +from operations_center.backends.limit_classifier import classify_limit +from operations_center.backends.worker_backend_selector import ( + maybe_record_worker_backend_cooldown, +) from operations_center.backends.factory import CanonicalBackendRegistry from operations_center.contracts.common import ValidationSummary from operations_center.contracts.enums import ( @@ -65,6 +70,20 @@ logger = logging.getLogger(__name__) +def _is_capacity_limit_failure(result: ExecutionResult) -> bool: + """True when ``result`` reflects quota/session exhaustion, not a code failure.""" + if ( + result.success + or result.failure_category != FailureReasonCategory.BACKEND_ERROR + or not result.failure_reason + ): + return False + if classify_capacity_exhaustion(result.failure_reason) is not None: + return True + limit_kind, _model = classify_limit(result.failure_reason) + return limit_kind is not None + + @runtime_checkable class _CaptureCapableAdapter(Protocol): def execute_and_capture(self, request) -> tuple[ExecutionResult, object | None]: ... @@ -335,27 +354,16 @@ def execute( repo_key=request.repo_key, backend=backend_name, ) + self._record_worker_backend_cooldown_from_result( + result=result, + runtime_metadata=runtime_metadata, + now=now, + ) # Capacity exhaustion is an external infrastructure event (rate-limited # API key, billing cap). Record it as a quota_event so it does NOT feed # the circuit breaker — the CB is a code/task-quality signal, not an # API-quota signal. See usage_store.record_quota_event docstring. - _is_capacity_exhaustion = ( - not result.success - and result.failure_category == FailureReasonCategory.BACKEND_ERROR - and result.failure_reason is not None - and any( - kw in result.failure_reason.lower() - for kw in ( - "capacity exhaustion", - "you've hit your limit", - "hit your limit", - "quota exceeded", - "rate limit", - "billing", - ) - ) - ) - if _is_capacity_exhaustion: + if _is_capacity_limit_failure(result): self._usage_store.record_quota_event( task_id=request.run_id, role=role, @@ -415,6 +423,30 @@ def execute( executed=True, ) + def _record_worker_backend_cooldown_from_result( + self, + *, + result: ExecutionResult, + runtime_metadata: dict[str, Any], + now: datetime, + ) -> None: + """Persist a worker-backend cooldown inferred from the final failure result.""" + if self._usage_store is None or not _is_capacity_limit_failure(result): + return + observed_runtime = runtime_metadata.get("observed_runtime") + if not isinstance(observed_runtime, dict): + return + worker_backend = observed_runtime.get("selected_worker_backend") + if not isinstance(worker_backend, str) or not worker_backend: + return + maybe_record_worker_backend_cooldown( + usage_store=self._usage_store, + worker_backend=worker_backend, + combined_output=result.failure_reason, + now=now, + logger=logger.info, + ) + def _apply_runtime_binding_policy( self, bundle: ProposalDecisionBundle, diff --git a/tests/unit/execution/test_coordinator_cov.py b/tests/unit/execution/test_coordinator_cov.py index a82eb705..6ce3e06c 100644 --- a/tests/unit/execution/test_coordinator_cov.py +++ b/tests/unit/execution/test_coordinator_cov.py @@ -77,6 +77,17 @@ def execute(self, request): return self.result +class _CaptureAdapter(_RecordingAdapter): + def __init__(self, result: ExecutionResult, capture) -> None: + super().__init__(result) + self.capture = capture + + def execute_and_capture(self, request): + self.calls += 1 + self.last_request = request + return self.result, self.capture + + class _CrashAdapter: def __init__(self) -> None: self.calls = 0 @@ -419,6 +430,7 @@ def __init__( self.events: list[str] = [] self.quota_events = 0 self.outcomes: list[bool] = [] + self.cooldowns: list[dict[str, object]] = [] self._global_conc = global_conc self._global_rate = global_rate self._global_mem = global_mem @@ -441,6 +453,9 @@ def record_quota_event(self, **_k): def record_execution_outcome(self, *, succeeded, **_k): self.outcomes.append(succeeded) + def record_worker_backend_cooldown(self, **kwargs): + self.cooldowns.append(kwargs) + def global_concurrency_decision(self, **_k): return self._global_conc @@ -664,6 +679,39 @@ def test_capacity_exhaustion_records_quota_event() -> None: assert store.outcomes == [] +def test_weekly_limit_records_worker_backend_cooldown_from_observed_runtime() -> None: + bundle = _bundle() + adapter = _CaptureAdapter( + _backend_failure(bundle, "You've hit your weekly limit · resets 9am (America/New_York)"), + capture=type( + "Capture", + (), + { + "observed_runtime": { + "preferred_worker_backend": "claude_code", + "selected_worker_backend": "claude_code", + "fallback_used": False, + } + }, + )(), + ) + store = _FakeUsageStore() + coord = ExecutionCoordinator( + adapter_registry=_Registry(adapter), + policy_engine=_AllowPolicy(), + usage_store=store, + ) + + out = coord.execute(bundle, _runtime()) + + assert out.executed is True + assert store.quota_events == 1 + assert store.outcomes == [] + assert len(store.cooldowns) == 1 + assert store.cooldowns[0]["worker_backend"] == "claude_code" + assert store.cooldowns[0]["limit_kind"] == "global_weekly" + + def test_non_capacity_failure_records_outcome() -> None: bundle = _bundle() adapter = _RecordingAdapter(_backend_failure(bundle, "syntax error in patch"))