diff --git a/README.md b/README.md index aa8bd5c2..c6405442 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-21) — Bulkhead & Rate-Limit Headers](#whats-new-2026-06-21--bulkhead--rate-limit-headers) - [What's new (2026-06-21) — Streaming Latency Percentiles](#whats-new-2026-06-21--streaming-latency-percentiles) - [What's new (2026-06-21) — Service-Level Objectives (SLO)](#whats-new-2026-06-21--service-level-objectives-slo) - [What's new (2026-06-21) — Chaos Experiments](#whats-new-2026-06-21--chaos-experiments) @@ -126,6 +127,12 @@ --- +## What's new (2026-06-21) — Bulkhead & Rate-Limit Headers + +Cap concurrency, honor server back-off. Full reference: [`docs/source/Eng/doc/new_features/v74_features_doc.rst`](docs/source/Eng/doc/new_features/v74_features_doc.rst). + +- **`Bulkhead` / `next_delay` / `parse_retry_after` / `parse_ratelimit`** (`AC_bulkhead_run`, `AC_retry_after`): `resilience` recovers and `rate_limit` paces, but nothing capped *simultaneous* in-flight calls (a slow dependency could exhaust every worker) and the HTTP client ignored `Retry-After`/`RateLimit-*`. This adds a bulkhead (bounded-concurrency permit that sheds load with `BulkheadFullError` when full) and parsers for the server's advised delay (delta-seconds or HTTP-date). Non-blocking permit counting → deterministic, no threads in tests. Pure-stdlib. + ## What's new (2026-06-21) — Streaming Latency Percentiles Mergeable p99 for load/soak runs. Full reference: [`docs/source/Eng/doc/new_features/v73_features_doc.rst`](docs/source/Eng/doc/new_features/v73_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index 5c33622f..e350bf0b 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-21) — 隔舱与速率限制标头](#本次更新-2026-06-21--隔舱与速率限制标头) - [本次更新 (2026-06-21) — 流式延迟百分位](#本次更新-2026-06-21--流式延迟百分位) - [本次更新 (2026-06-21) — 服务等级目标(SLO)](#本次更新-2026-06-21--服务等级目标slo) - [本次更新 (2026-06-21) — 混沌实验](#本次更新-2026-06-21--混沌实验) @@ -125,6 +126,12 @@ --- +## 本次更新 (2026-06-21) — 隔舱与速率限制标头 + +限制并发、遵守服务器退避。完整参考:[`docs/source/Zh/doc/new_features/v74_features_doc.rst`](../docs/source/Zh/doc/new_features/v74_features_doc.rst)。 + +- **`Bulkhead` / `next_delay` / `parse_retry_after` / `parse_ratelimit`**(`AC_bulkhead_run`、`AC_retry_after`):`resilience` 恢复、`rate_limit` 调速,但没有任何东西限制*同时*进行的调用(缓慢依赖会耗尽所有 worker),且 HTTP 客户端忽略 `Retry-After`/`RateLimit-*`。本功能补上隔舱(满载时以 `BulkheadFullError` 卸除负载的 bounded-concurrency 许可)以及服务器建议延迟(delta 秒或 HTTP-date)的解析器。非阻塞许可计数 → 确定、测试免线程。纯标准库。 + ## 本次更新 (2026-06-21) — 流式延迟百分位 load/soak 测试的可合并 p99。完整参考:[`docs/source/Zh/doc/new_features/v73_features_doc.rst`](../docs/source/Zh/doc/new_features/v73_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 181defc8..eeb11f68 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-21) — 隔艙與速率限制標頭](#本次更新-2026-06-21--隔艙與速率限制標頭) - [本次更新 (2026-06-21) — 串流延遲百分位](#本次更新-2026-06-21--串流延遲百分位) - [本次更新 (2026-06-21) — 服務等級目標(SLO)](#本次更新-2026-06-21--服務等級目標slo) - [本次更新 (2026-06-21) — 混沌實驗](#本次更新-2026-06-21--混沌實驗) @@ -125,6 +126,12 @@ --- +## 本次更新 (2026-06-21) — 隔艙與速率限制標頭 + +限制並行、遵守伺服器退避。完整參考:[`docs/source/Zh/doc/new_features/v74_features_doc.rst`](../docs/source/Zh/doc/new_features/v74_features_doc.rst)。 + +- **`Bulkhead` / `next_delay` / `parse_retry_after` / `parse_ratelimit`**(`AC_bulkhead_run`、`AC_retry_after`):`resilience` 復原、`rate_limit` 調速,但沒有任何東西限制*同時*進行的呼叫(緩慢相依會耗盡所有 worker),且 HTTP 用戶端忽略 `Retry-After`/`RateLimit-*`。本功能補上隔艙(滿載時以 `BulkheadFullError` 卸除負載的 bounded-concurrency 許可)以及伺服器建議延遲(delta 秒或 HTTP-date)的剖析器。非阻塞許可計數 → 具決定性、測試免執行緒。純標準函式庫。 + ## 本次更新 (2026-06-21) — 串流延遲百分位 load/soak 測試的可合併 p99。完整參考:[`docs/source/Zh/doc/new_features/v73_features_doc.rst`](../docs/source/Zh/doc/new_features/v73_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v74_features_doc.rst b/docs/source/Eng/doc/new_features/v74_features_doc.rst new file mode 100644 index 00000000..adfdc51b --- /dev/null +++ b/docs/source/Eng/doc/new_features/v74_features_doc.rst @@ -0,0 +1,48 @@ +Bulkhead & Rate-Limit Headers +============================= + +``resilience`` recovers from failures and ``rate_limit`` paces calls, but +nothing capped the number of *simultaneous* in-flight calls to one resource +(so a slow dependency could exhaust every worker), and the HTTP client read +``Retry-After`` / ``RateLimit-*`` response headers but honored none of them. +This adds a bulkhead (bounded-concurrency permit with load-shedding) and a +parser for the server's advised delay. + +Pure standard library (``threading`` + ``email.utils``); the permit counting is +non-blocking (reject when full), so it is deterministic and CI-testable without +spawning threads. Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import Bulkhead, BulkheadFullError, next_delay + + payments = Bulkhead(max_concurrent=4, name="payments") + try: + result = payments.run(call_payment_api, order) + except BulkheadFullError: + defer(order) # shed load instead of piling on + + # honor the server's back-off after an HTTP call + wait = next_delay(response) # from Retry-After / RateLimit-* headers + if wait: + sleep(wait) + +``Bulkhead`` caps simultaneous holders to ``max_concurrent`` — ``try_enter`` / +``release``, a context manager, and ``run(func)`` all reject (``BulkheadFullError``) +when full, isolating one slow dependency from exhausting the rest. +``parse_retry_after`` understands both the delta-seconds and HTTP-date forms; +``parse_ratelimit`` reads the ``RateLimit-Limit/Remaining/Reset`` convention; +``next_delay`` combines them into the wait a flow should observe after a +``429`` / ``503``. + +Executor commands +----------------- + +``AC_bulkhead_run`` runs an ``actions`` list under a named bulkhead (``name``, +``max_concurrent``) and returns ``{entered, in_flight, record?}``. +``AC_retry_after`` takes an HTTP ``response`` (``{status, headers}``) and returns +``{delay}``. Both are exposed as MCP tools (``ac_bulkhead_run`` / +``ac_retry_after``) and as Script Builder commands under **Flow**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 6280667b..9b1c3b06 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -96,6 +96,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v71_features_doc doc/new_features/v72_features_doc doc/new_features/v73_features_doc + doc/new_features/v74_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v74_features_doc.rst b/docs/source/Zh/doc/new_features/v74_features_doc.rst new file mode 100644 index 00000000..c45b094a --- /dev/null +++ b/docs/source/Zh/doc/new_features/v74_features_doc.rst @@ -0,0 +1,42 @@ +隔艙與速率限制標頭 +================= + +``resilience`` 從失敗中復原,``rate_limit`` 為呼叫調速,但沒有任何東西能限制對單一資源*同時*進行 +的呼叫數(因此一個緩慢的相依會耗盡所有 worker),而 HTTP 用戶端會讀取 ``Retry-After`` / +``RateLimit-*`` 回應標頭卻不予理會。本功能補上一個隔艙(bounded-concurrency 許可,含負載卸除) +以及一個解析伺服器建議延遲的剖析器。 + +純標準函式庫(``threading`` + ``email.utils``);許可計數為非阻塞(滿載即拒絕),因此具決定性、 +不需開執行緒即可在 CI 測試。不匯入 ``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import Bulkhead, BulkheadFullError, next_delay + + payments = Bulkhead(max_concurrent=4, name="payments") + try: + result = payments.run(call_payment_api, order) + except BulkheadFullError: + defer(order) # 卸除負載而非繼續堆積 + + # HTTP 呼叫後遵守伺服器的退避 + wait = next_delay(response) # 來自 Retry-After / RateLimit-* 標頭 + if wait: + sleep(wait) + +``Bulkhead`` 把同時持有者上限設為 ``max_concurrent`` —— ``try_enter`` / ``release``、context +manager 與 ``run(func)`` 在滿載時皆拒絕(``BulkheadFullError``),把一個緩慢相依與其餘隔離。 +``parse_retry_after`` 同時理解 delta 秒數與 HTTP-date 兩種形式;``parse_ratelimit`` 讀取 +``RateLimit-Limit/Remaining/Reset`` 慣例;``next_delay`` 把它們結合成流程在 ``429`` / ``503`` 後 +應遵守的等待。 + +執行器命令 +---------- + +``AC_bulkhead_run`` 在具名隔艙(``name``、``max_concurrent``)下執行 ``actions`` 清單,回傳 +``{entered, in_flight, record?}``。``AC_retry_after`` 接受 HTTP ``response``(``{status, headers}``) +回傳 ``{delay}``。兩者皆以 MCP 工具(``ac_bulkhead_run`` / ``ac_retry_after``)以及 Script Builder +中 **Flow** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 70998c70..d2c21a42 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -96,6 +96,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v71_features_doc doc/new_features/v72_features_doc doc/new_features/v73_features_doc + doc/new_features/v74_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 957dcee0..bdde04a7 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -371,6 +371,10 @@ ) # Mergeable streaming latency digest + exact percentiles from je_auto_control.utils.percentiles import LatencyDigest, exact_percentiles +# Bulkhead concurrency isolation + rate-limit header parsing +from je_auto_control.utils.bulkhead import ( + Bulkhead, BulkheadFullError, next_delay, parse_ratelimit, parse_retry_after, +) # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -868,6 +872,8 @@ def start_autocontrol_gui(*args, **kwargs): "run_experiment", "BurnRule", "burn_alerts", "burn_rate", "default_burn_rules", "evaluate_slo", "LatencyDigest", "exact_percentiles", + "Bulkhead", "BulkheadFullError", "next_delay", "parse_ratelimit", + "parse_retry_after", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index b8711acf..293df940 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -1575,6 +1575,24 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None: ), description="Run 'actions' (JSON view) via a named circuit breaker.", )) + specs.append(CommandSpec( + "AC_bulkhead_run", "Flow", "Bulkhead (Bounded Concurrency)", + fields=( + FieldSpec("name", FieldType.STRING), + FieldSpec("max_concurrent", FieldType.INT, default=4), + FieldSpec("actions", FieldType.STRING, + placeholder="[[\"AC_click_mouse\", {...}]]"), + ), + description="Run 'actions' (JSON view) under a named bulkhead permit.", + )) + specs.append(CommandSpec( + "AC_retry_after", "Flow", "Parse Retry-After / RateLimit", + fields=( + FieldSpec("response", FieldType.STRING, + placeholder='{"status": 429, "headers": {"Retry-After": "30"}}'), + ), + description="Server-advised wait (seconds) from an HTTP response; {delay}.", + )) specs.append(CommandSpec( "AC_rate_limit", "Flow", "Rate Limit (Token Bucket)", fields=( diff --git a/je_auto_control/utils/bulkhead/__init__.py b/je_auto_control/utils/bulkhead/__init__.py new file mode 100644 index 00000000..95a99c0c --- /dev/null +++ b/je_auto_control/utils/bulkhead/__init__.py @@ -0,0 +1,9 @@ +"""Bulkhead concurrency isolation + server rate-limit header parsing.""" +from je_auto_control.utils.bulkhead.bulkhead import ( + Bulkhead, BulkheadFullError, next_delay, parse_ratelimit, parse_retry_after, +) + +__all__ = [ + "Bulkhead", "BulkheadFullError", "next_delay", "parse_ratelimit", + "parse_retry_after", +] diff --git a/je_auto_control/utils/bulkhead/bulkhead.py b/je_auto_control/utils/bulkhead/bulkhead.py new file mode 100644 index 00000000..2772ed9e --- /dev/null +++ b/je_auto_control/utils/bulkhead/bulkhead.py @@ -0,0 +1,125 @@ +"""Bulkhead concurrency isolation + server rate-limit header parsing. + +``resilience`` recovers from failures and ``rate_limit`` paces calls, but +nothing capped the number of *simultaneous* in-flight calls to one resource +(so a slow dependency could exhaust every worker), and the HTTP client read +``Retry-After`` / ``RateLimit-*`` headers but honored none of them. This adds a +bulkhead (bounded-concurrency permit with load-shedding) and a parser for the +server's advised delay. + +The permit counting is lock-guarded and non-blocking (reject when full), so the +accounting is deterministic and CI-testable without spawning threads. Pure +standard library (``threading`` + ``email.utils``); imports no ``PySide6``. +""" +import threading +import time +from email.utils import parsedate_to_datetime +from typing import Any, Callable, Dict, Mapping, Optional + +from je_auto_control.utils.exception.exceptions import AutoControlException + + +class BulkheadFullError(AutoControlException): + """Raised when a bulkhead has no free permit.""" + + +class Bulkhead: + """Cap simultaneous in-flight calls to one resource; reject when full.""" + + def __init__(self, max_concurrent: int, *, name: str = "bulkhead") -> None: + if max_concurrent < 1: + raise AutoControlException("max_concurrent must be >= 1") + self.name = name + self._max = int(max_concurrent) + self._in_flight = 0 + self._lock = threading.Lock() + + @property + def in_flight(self) -> int: + """Current number of held permits.""" + with self._lock: + return self._in_flight + + def try_enter(self) -> bool: + """Take a permit if one is free; return whether it succeeded.""" + with self._lock: + if self._in_flight < self._max: + self._in_flight += 1 + return True + return False + + def release(self) -> None: + """Return a permit.""" + with self._lock: + if self._in_flight > 0: + self._in_flight -= 1 + + def __enter__(self) -> "Bulkhead": + if not self.try_enter(): + raise BulkheadFullError(f"{self.name} is full ({self._max})") + return self + + def __exit__(self, *_exc: Any) -> bool: + self.release() + return False + + def run(self, func: Callable, *args: Any, **kwargs: Any) -> Any: + """Run ``func`` under a permit (raises ``BulkheadFullError`` if full).""" + with self: + return func(*args, **kwargs) + + +def _get_header(headers: Mapping[str, Any], name: str) -> Optional[Any]: + lower = name.lower() + for key, value in headers.items(): + if key.lower() == lower: + return value + return None + + +def parse_retry_after(headers: Mapping[str, Any], *, + now: Optional[float] = None) -> Optional[float]: + """Return the ``Retry-After`` delay in seconds (delta or HTTP-date).""" + raw = _get_header(headers, "Retry-After") + if raw is None: + return None + text = str(raw).strip() + if text.isdigit(): + return float(text) + try: + target = parsedate_to_datetime(text) + except (TypeError, ValueError): + return None + when = time.time() if now is None else now + return max(0.0, target.timestamp() - when) + + +def _as_int(value: Any) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def parse_ratelimit(headers: Mapping[str, Any]) -> Optional[Dict[str, Any]]: + """Parse de-facto ``RateLimit-Limit/Remaining/Reset`` headers.""" + limit = _get_header(headers, "RateLimit-Limit") + remaining = _get_header(headers, "RateLimit-Remaining") + reset = _get_header(headers, "RateLimit-Reset") + if limit is None and remaining is None and reset is None: + return None + return {"limit": _as_int(limit), "remaining": _as_int(remaining), + "reset": _as_int(reset)} + + +def next_delay(response: Mapping[str, Any], *, + now: Optional[float] = None) -> float: + """Return the server-advised wait (seconds) from a response, or 0.0.""" + headers = response.get("headers", {}) if isinstance(response, Mapping) else {} + delay = parse_retry_after(headers, now=now) + if delay is not None: + return delay + info = parse_ratelimit(headers) + if info and info.get("remaining") == 0 and info.get("reset") is not None: + return float(info["reset"]) + return 0.0 diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 70beca09..dc113258 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2928,6 +2928,35 @@ def _rate_limit(name: str, rate: float = 1.0, capacity: float = 1.0, "wait": round(bucket.time_until_available(float(n)), 4)} +_BULKHEADS: Dict[str, Any] = {} + + +def _bulkhead_run(name: str, max_concurrent: int, + actions: Any) -> Dict[str, Any]: + """Adapter: run an action list under a named bulkhead permit.""" + import json + from je_auto_control.utils.bulkhead import Bulkhead, BulkheadFullError + if isinstance(actions, str): + actions = json.loads(actions) + bulkhead = _BULKHEADS.setdefault( + name, Bulkhead(int(max_concurrent), name=name)) + try: + with bulkhead: + record = executor.execute_action(list(actions), raise_on_error=True) + except BulkheadFullError: + return {"entered": False, "in_flight": bulkhead.in_flight} + return {"entered": True, "in_flight": bulkhead.in_flight, "record": record} + + +def _retry_after(response: Any) -> Dict[str, Any]: + """Adapter: server-advised wait from a response (dict or JSON string).""" + import json + from je_auto_control.utils.bulkhead import next_delay + if isinstance(response, str): + response = json.loads(response) + return {"delay": next_delay(response)} + + def _percentiles(samples: Any, qs: Any = None) -> Dict[str, Any]: """Adapter: exact percentiles of a numeric sample list (or JSON string).""" import json @@ -4001,6 +4030,8 @@ def __init__(self): "AC_evaluate_slo": _evaluate_slo, "AC_burn_alerts": _burn_alerts, "AC_percentiles": _percentiles, + "AC_bulkhead_run": _bulkhead_run, + "AC_retry_after": _retry_after, "AC_unified_diff": _unified_diff, "AC_apply_unified": _apply_unified, "AC_three_way_merge": _three_way_merge, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index b76e7a19..1b0a28d7 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3343,6 +3343,33 @@ def rate_limit_tools() -> List[MCPTool]: ] +def bulkhead_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_bulkhead_run", + description=("Run an 'actions' list under a named bulkhead permit " + "(max 'max_concurrent' in-flight; rejects when full). " + "Returns {entered, in_flight, record?}."), + input_schema=schema( + {"name": {"type": "string"}, + "max_concurrent": {"type": "integer"}, + "actions": {"type": "array"}}, + ["name", "max_concurrent", "actions"]), + handler=h.bulkhead_run, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_retry_after", + description=("Server-advised wait (seconds) from an HTTP 'response' " + "{status, headers} via Retry-After / RateLimit-*. " + "Returns {delay}."), + input_schema=schema({"response": {"type": "object"}}, ["response"]), + handler=h.retry_after, + annotations=READ_ONLY, + ), + ] + + def percentiles_tools() -> List[MCPTool]: return [ MCPTool( @@ -4833,7 +4860,7 @@ def media_assert_tools() -> List[MCPTool]: license_policy_tools, jwt_tools, rate_limit_tools, json_patch_tools, search_index_tools, stats_tools, recurrence_tools, text_diff_tools, feature_flag_tools, provenance_tools, json_contract_tools, chaos_tools, - slo_tools, percentiles_tools, + slo_tools, percentiles_tools, bulkhead_tools, saga_tools, decision_table_tools, locator_repair_tools, pii_text_tools, sarif_tools, screen_record_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 4bf4b774..062bb385 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1697,6 +1697,16 @@ def percentiles(samples, qs=None): return {"percentiles": {str(q): value for q, value in result.items()}} +def bulkhead_run(name, max_concurrent, actions): + from je_auto_control.utils.executor.action_executor import _bulkhead_run + return _bulkhead_run(name, max_concurrent, actions) + + +def retry_after(response): + from je_auto_control.utils.bulkhead import next_delay + return {"delay": next_delay(response)} + + def build_provenance(paths, builder_id="je_auto_control"): from je_auto_control.utils.provenance import build_provenance, subject_for subjects = [subject_for(path) for path in paths] diff --git a/test/unit_test/headless/test_bulkhead_batch.py b/test/unit_test/headless/test_bulkhead_batch.py new file mode 100644 index 00000000..e066e473 --- /dev/null +++ b/test/unit_test/headless/test_bulkhead_batch.py @@ -0,0 +1,95 @@ +"""Headless tests for the bulkhead + rate-limit header parser. Pure stdlib.""" +import json + +import pytest + +import je_auto_control as ac +from je_auto_control.utils.bulkhead import ( + Bulkhead, BulkheadFullError, next_delay, parse_ratelimit, parse_retry_after) +from je_auto_control.utils.exception.exceptions import AutoControlException + + +def test_bulkhead_permits_then_rejects(): + bulkhead = Bulkhead(2, name="api") + assert bulkhead.try_enter() is True + assert bulkhead.try_enter() is True + assert bulkhead.try_enter() is False # full + assert bulkhead.in_flight == 2 + bulkhead.release() + assert bulkhead.try_enter() is True + + +def test_bulkhead_context_manager_rejects_when_full(): + bulkhead = Bulkhead(1) + with bulkhead: + assert bulkhead.try_enter() is False # full + with pytest.raises(BulkheadFullError): + bulkhead.run(lambda: "unreached") # rejects on enter + assert bulkhead.in_flight == 0 # released after context + + +def test_bulkhead_run_and_bad_max(): + assert Bulkhead(1).run(lambda x: x * 2, 21) == 42 + with pytest.raises(AutoControlException): + Bulkhead(0) + + +def test_retry_after_delta_and_date_and_case(): + assert parse_retry_after({"Retry-After": "120"}) == pytest.approx(120.0) + assert parse_retry_after({"retry-after": "5"}) == pytest.approx(5.0) + future = parse_retry_after({"Retry-After": "Wed, 21 Oct 2099 07:28:00 GMT"}, + now=0) + assert future > 0 + assert parse_retry_after({"X": "1"}) is None + + +def test_parse_ratelimit(): + info = parse_ratelimit({"RateLimit-Limit": "100", "RateLimit-Remaining": "0", + "RateLimit-Reset": "30"}) + assert info == {"limit": 100, "remaining": 0, "reset": 30} + assert parse_ratelimit({"X": "1"}) is None + + +def test_next_delay_sources(): + assert next_delay({"status": 429, "headers": {"Retry-After": "10"}}) == \ + pytest.approx(10.0) + assert next_delay({"headers": {"RateLimit-Remaining": "0", + "RateLimit-Reset": "15"}}) == pytest.approx(15.0) + assert next_delay({"headers": {}}) == pytest.approx(0.0) + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(): + rec = ac.execute_action([[ + "AC_retry_after", + {"response": json.dumps({"status": 429, + "headers": {"Retry-After": "42"}})}, + ]]) + assert next(v for v in rec.values() if isinstance(v, dict))["delay"] == \ + pytest.approx(42.0) + + rec2 = ac.execute_action([[ + "AC_bulkhead_run", + {"name": "t", "max_concurrent": 2, + "actions": json.dumps([["AC_sleep", {"seconds": 0}]])}, + ]]) + payload = next(v for v in rec2.values() if isinstance(v, dict)) + assert payload["entered"] is True + + +def test_wiring(): + assert {"AC_bulkhead_run", "AC_retry_after"} <= ac.executor.known_commands() + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert {"ac_bulkhead_run", "ac_retry_after"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_bulkhead_run", "AC_retry_after"} <= cmds + + +def test_facade_exports(): + for attr in ("Bulkhead", "BulkheadFullError", "next_delay", + "parse_ratelimit", "parse_retry_after"): + assert hasattr(ac, attr) + assert attr in ac.__all__