Merge pull request #282 from Integration-Automation/feat/bulkhead

JE-Chen · web-flow · commit de8b4bf5ce63 · 2026-06-21T21:10:48.000+08:00
Add bulkhead concurrency isolation and rate-limit header parsing
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py
diff --git a/test/unit_test/headless/test_bulkhead_batch.py b/test/unit_test/headless/test_bulkhead_batch.py
