fix(OperationsCenter): persist worker backend cooldowns from execution failures

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,

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"))