Add mergeable streaming latency percentile digest

README.md

@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [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)
 - [What's new (2026-06-21) — JSON Contract & Snapshot Matching](#whats-new-2026-06-21--json-contract--snapshot-matching)
@@ -125,6 +126,12 @@
 
 ---
 
+## 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).
+
+- **`LatencyDigest` / `exact_percentiles`** (`AC_percentiles`): `stats.percentile` needs the full sorted list; this adds a HdrHistogram-style digest with O(1) `record`, bounded memory (significant-figure buckets), and `merge` for cross-shard aggregation — the property you need for a correct aggregate p99 from per-worker results. `exact_percentiles` covers the small-set case (arbitrary quantiles). Pure-stdlib `math`.
+
 ## What's new (2026-06-21) — Service-Level Objectives (SLO)
 
 SLI, error budget and burn-rate alerts. Full reference: [`docs/source/Eng/doc/new_features/v72_features_doc.rst`](docs/source/Eng/doc/new_features/v72_features_doc.rst).

README/README_zh-CN.md

@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-21) — 流式延迟百分位](#本次更新-2026-06-21--流式延迟百分位)
 - [本次更新 (2026-06-21) — 服务等级目标(SLO)](#本次更新-2026-06-21--服务等级目标slo)
 - [本次更新 (2026-06-21) — 混沌实验](#本次更新-2026-06-21--混沌实验)
 - [本次更新 (2026-06-21) — JSON 合约与快照比对](#本次更新-2026-06-21--json-合约与快照比对)
@@ -124,6 +125,12 @@
 
 ---
 
+## 本次更新 (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)。
+
+- **`LatencyDigest` / `exact_percentiles`**(`AC_percentiles`):`stats.percentile` 需要完整已排序列表;本功能补上 HdrHistogram 风格的 digest,具 O(1) `record`、内存有界(有效位数分桶)以及跨分片汇聚的 `merge` —— 这正是从各 worker 结果计算正确汇聚 p99 所需的特性。`exact_percentiles` 涵盖小样本集情况(任意分位)。纯标准库 `math`。
+
 ## 本次更新 (2026-06-21) — 服务等级目标(SLO)
 
 SLI、错误预算与燃烧率告警。完整参考:[`docs/source/Zh/doc/new_features/v72_features_doc.rst`](../docs/source/Zh/doc/new_features/v72_features_doc.rst)。

README/README_zh-TW.md

@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-21) — 串流延遲百分位](#本次更新-2026-06-21--串流延遲百分位)
 - [本次更新 (2026-06-21) — 服務等級目標(SLO)](#本次更新-2026-06-21--服務等級目標slo)
 - [本次更新 (2026-06-21) — 混沌實驗](#本次更新-2026-06-21--混沌實驗)
 - [本次更新 (2026-06-21) — JSON 合約與快照比對](#本次更新-2026-06-21--json-合約與快照比對)
@@ -124,6 +125,12 @@
 
 ---
 
+## 本次更新 (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)。
+
+- **`LatencyDigest` / `exact_percentiles`**(`AC_percentiles`):`stats.percentile` 需要完整已排序清單;本功能補上 HdrHistogram 風格的 digest,具 O(1) `record`、記憶體有界(有效位數分桶)以及跨分片彙整的 `merge` —— 這正是從各 worker 結果計算正確彙整 p99 所需的特性。`exact_percentiles` 涵蓋小樣本集情況(任意分位)。純標準函式庫 `math`。
+
 ## 本次更新 (2026-06-21) — 服務等級目標(SLO)
 
 SLI、錯誤預算與燃燒率警示。完整參考:[`docs/source/Zh/doc/new_features/v72_features_doc.rst`](../docs/source/Zh/doc/new_features/v72_features_doc.rst)。

docs/source/Eng/doc/new_features/v73_features_doc.rst

@@ -0,0 +1,43 @@
+Streaming Latency Percentiles
+=============================
+
+``stats.percentile`` is exact but needs the full sorted sample list in memory;
+for a long-running or sharded load / soak run you want an O(1)-per-record,
+bounded-memory, *mergeable* structure instead. This adds a HdrHistogram-style
+:class:`LatencyDigest` (records into significant-figure buckets, merges across
+shards) plus :func:`exact_percentiles` for small sample sets.
+
+Pure standard library (``math``); imports no ``PySide6``.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import LatencyDigest, exact_percentiles
+
+    digest = LatencyDigest(sig_figs=3)
+    for latency_ms in stream:
+        digest.record(latency_ms)        # O(1), bounded memory
+    print(digest.summary())              # min/mean/max/p50/p90/p95/p99
+
+    # merge per-shard digests into one
+    total = shard_a.merge(shard_b)
+
+    # exact percentiles for a small in-memory set
+    exact_percentiles([12.0, 9.5, 14.2], qs=(50, 95))
+
+``LatencyDigest.record`` buckets each value to ``sig_figs`` significant figures
+(so memory is bounded by the number of distinct rounded values, not the sample
+count); ``percentile`` / ``quantiles`` / ``summary`` read it back, and ``merge``
+folds another digest in for cross-shard aggregation — the property you need to
+compute a correct aggregate p99 from per-worker results. ``exact_percentiles``
+delegates to ``stats.percentile`` for the small-set case.
+
+Executor command
+----------------
+
+``AC_percentiles`` takes ``samples`` (a list or JSON string) and optional ``qs``
+quantiles (default 50/90/95/99) and returns ``{percentiles}``. The same
+operation is exposed as the MCP tool ``ac_percentiles`` and as a Script Builder
+command under **Report**.

docs/source/Eng/eng_index.rst

@@ -95,6 +95,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v70_features_doc
    doc/new_features/v71_features_doc
    doc/new_features/v72_features_doc
+   doc/new_features/v73_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

docs/source/Zh/doc/new_features/v73_features_doc.rst

@@ -0,0 +1,39 @@
+串流延遲百分位
+=============
+
+``stats.percentile`` 精確,但需要把整份已排序的樣本清單放在記憶體中;對長時間執行或分片的
+load / soak 測試,你會想要一個每筆 O(1)、記憶體有界、且**可合併**的結構。本功能補上 HdrHistogram
+風格的 :class:`LatencyDigest`(以有效位數分桶記錄、可跨分片合併),外加供小樣本集使用的
+:func:`exact_percentiles`。
+
+純標準函式庫(``math``);不匯入 ``PySide6``。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import LatencyDigest, exact_percentiles
+
+    digest = LatencyDigest(sig_figs=3)
+    for latency_ms in stream:
+        digest.record(latency_ms)        # O(1)、記憶體有界
+    print(digest.summary())              # min/mean/max/p50/p90/p95/p99
+
+    # 把各分片的 digest 合併成一個
+    total = shard_a.merge(shard_b)
+
+    # 小樣本集的精確百分位
+    exact_percentiles([12.0, 9.5, 14.2], qs=(50, 95))
+
+``LatencyDigest.record`` 把每個值四捨五入到 ``sig_figs`` 有效位數分桶(因此記憶體由相異捨入值的
+數量決定,而非樣本數);``percentile`` / ``quantiles`` / ``summary`` 讀回,而 ``merge`` 把另一個
+digest 折入以做跨分片彙整 —— 這正是從各 worker 結果計算正確彙整 p99 所需的特性。
+``exact_percentiles`` 在小樣本集情況下委派給 ``stats.percentile``。
+
+執行器命令
+----------
+
+``AC_percentiles`` 接受 ``samples``(清單或 JSON 字串)與選用的 ``qs`` 分位(預設 50/90/95/99),
+回傳 ``{percentiles}``。同一操作亦以 MCP 工具 ``ac_percentiles`` 以及 Script Builder 中
+**Report** 分類下的命令提供。

docs/source/Zh/zh_index.rst

@@ -95,6 +95,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v70_features_doc
    doc/new_features/v71_features_doc
    doc/new_features/v72_features_doc
+   doc/new_features/v73_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

je_auto_control/__init__.py

@@ -369,6 +369,8 @@
 from je_auto_control.utils.slo import (
     BurnRule, burn_alerts, burn_rate, default_burn_rules, evaluate_slo,
 )
+# Mergeable streaming latency digest + exact percentiles
+from je_auto_control.utils.percentiles import LatencyDigest, exact_percentiles
 # Background popup/interrupt watchdog (unattended automation)
 from je_auto_control.utils.watchdog import (
     PopupWatchdog, WatchdogRule, default_popup_watchdog,
@@ -865,6 +867,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "ChaosExperiment", "Fault", "Probe", "exception_fault", "latency_fault",
     "run_experiment",
     "BurnRule", "burn_alerts", "burn_rate", "default_burn_rules", "evaluate_slo",
+    "LatencyDigest", "exact_percentiles",
     # MCP server
     "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt",
     "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool",

je_auto_control/gui/script_builder/command_schema.py

@@ -1388,6 +1388,16 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Verify a JWT (alg allowlist + exp/nbf/aud); returns {ok, claims}.",
     ))
+    specs.append(CommandSpec(
+        "AC_percentiles", "Report", "Percentiles",
+        fields=(
+            FieldSpec("samples", FieldType.STRING,
+                      placeholder="[12.0, 9.5, 14.2, 11.1]"),
+            FieldSpec("qs", FieldType.STRING, optional=True,
+                      placeholder="[50, 90, 99]"),
+        ),
+        description="Exact percentiles of a numeric sample list.",
+    ))
     specs.append(CommandSpec(
         "AC_evaluate_slo", "Report", "SLO: Evaluate (SLI + Error Budget)",
         fields=(

je_auto_control/utils/executor/action_executor.py

@@ -2928,6 +2928,19 @@ def _rate_limit(name: str, rate: float = 1.0, capacity: float = 1.0,
             "wait": round(bucket.time_until_available(float(n)), 4)}
 
 
+def _percentiles(samples: Any, qs: Any = None) -> Dict[str, Any]:
+    """Adapter: exact percentiles of a numeric sample list (or JSON string)."""
+    import json
+    from je_auto_control.utils.percentiles import exact_percentiles
+    if isinstance(samples, str):
+        samples = json.loads(samples)
+    if isinstance(qs, str):
+        qs = json.loads(qs)
+    quantiles = tuple(qs) if qs else (50, 90, 95, 99)
+    result = exact_percentiles(samples, qs=quantiles)
+    return {"percentiles": {str(q): value for q, value in result.items()}}
+
+
 def _evaluate_slo(records: Any, target: float,
                   window_s: Optional[float] = None) -> Dict[str, Any]:
     """Adapter: SLI + error budget for outcome records (list or JSON string)."""
@@ -3987,6 +4000,7 @@ def __init__(self):
             "AC_run_chaos": _run_chaos,
             "AC_evaluate_slo": _evaluate_slo,
             "AC_burn_alerts": _burn_alerts,
+            "AC_percentiles": _percentiles,
             "AC_unified_diff": _unified_diff,
             "AC_apply_unified": _apply_unified,
             "AC_three_way_merge": _three_way_merge,

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -3343,6 +3343,22 @@ def rate_limit_tools() -> List[MCPTool]:
     ]
 
 
+def percentiles_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_percentiles",
+            description=("Exact percentiles of a numeric 'samples' list at the "
+                         "requested 'qs' quantiles (default 50/90/95/99). "
+                         "Returns {percentiles}."),
+            input_schema=schema(
+                {"samples": {"type": "array"}, "qs": {"type": "array"}},
+                ["samples"]),
+            handler=h.percentiles,
+            annotations=READ_ONLY,
+        ),
+    ]
+
+
 def slo_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -4817,7 +4833,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,
+    slo_tools, percentiles_tools,
     saga_tools, decision_table_tools, locator_repair_tools,
     pii_text_tools, sarif_tools,
     screen_record_tools,

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -1691,6 +1691,12 @@ def burn_alerts(records, target):
     return {"alerts": alerts, "firing": bool(alerts)}
 
 
+def percentiles(samples, qs=None):
+    from je_auto_control.utils.percentiles import exact_percentiles
+    result = exact_percentiles(samples, qs=tuple(qs) if qs else (50, 90, 95, 99))
+    return {"percentiles": {str(q): value for q, value in result.items()}}
+
+
 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]

je_auto_control/utils/percentiles/__init__.py

@@ -0,0 +1,6 @@
+"""Mergeable streaming latency digest + exact percentiles."""
+from je_auto_control.utils.percentiles.percentiles import (
+    LatencyDigest, exact_percentiles,
+)
+
+__all__ = ["LatencyDigest", "exact_percentiles"]

je_auto_control/utils/percentiles/percentiles.py

@@ -0,0 +1,97 @@
+"""A mergeable streaming latency digest for percentile estimation.
+
+``stats.percentile`` is exact but needs the full sorted sample list in memory;
+for a long-running or sharded load/soak run you want an O(1)-per-record,
+bounded-memory, *mergeable* structure instead. This adds a HdrHistogram-style
+:class:`LatencyDigest` (records into significant-figure buckets, merges across
+shards) plus :func:`exact_percentiles` for small sample sets.
+
+Pure standard library (``math``); imports no ``PySide6``.
+"""
+import math
+from typing import Dict, Iterable, Optional, Sequence
+
+from je_auto_control.utils.stats import percentile
+
+
+def exact_percentiles(samples: Sequence[float],
+                      qs: Iterable[float] = (50, 90, 95, 99)) -> Dict[float, float]:
+    """Return exact percentiles for a small sample set (delegates to stats)."""
+    return {q: percentile(samples, q) for q in qs}
+
+
+class LatencyDigest:
+    """A mergeable percentile estimator using significant-figure buckets."""
+
+    def __init__(self, *, sig_figs: int = 3) -> None:
+        if sig_figs < 1:
+            raise ValueError("sig_figs must be >= 1")
+        self._sig = int(sig_figs)
+        self._counts: Dict[float, int] = {}
+        self._count = 0
+        self._sum = 0.0
+        self._min: Optional[float] = None
+        self._max: Optional[float] = None
+
+    def _bucket(self, value: float) -> float:
+        if value <= 0:
+            return 0.0
+        digits = self._sig - 1 - math.floor(math.log10(value))
+        return round(value, digits)
+
+    def record(self, value: float) -> None:
+        """Record one observation."""
+        value = float(value)
+        bucket = self._bucket(value)
+        self._counts[bucket] = self._counts.get(bucket, 0) + 1
+        self._count += 1
+        self._sum += value
+        self._min = value if self._min is None else min(self._min, value)
+        self._max = value if self._max is None else max(self._max, value)
+
+    @property
+    def count(self) -> int:
+        """Number of recorded observations."""
+        return self._count
+
+    def percentile(self, q: float) -> float:
+        """Return the estimated ``q``-th percentile (0-100)."""
+        if self._count == 0:
+            return 0.0
+        rank = (max(0.0, min(100.0, q)) / 100.0) * self._count
+        cumulative = 0
+        for bucket in sorted(self._counts):
+            cumulative += self._counts[bucket]
+            if cumulative >= rank:
+                return bucket
+        return self._max if self._max is not None else 0.0
+
+    def quantiles(self, qs: Iterable[float]) -> Dict[float, float]:
+        """Return ``{q: percentile(q)}`` for each requested quantile."""
+        return {q: self.percentile(q) for q in qs}
+
+    def summary(self) -> Dict[str, float]:
+        """Return min/mean/max/count and the standard tail percentiles."""
+        mean = self._sum / self._count if self._count else 0.0
+        return {
+            "count": self._count,
+            "min": self._min if self._min is not None else 0.0,
+            "max": self._max if self._max is not None else 0.0,
+            "mean": mean,
+            "p50": self.percentile(50), "p90": self.percentile(90),
+            "p95": self.percentile(95), "p99": self.percentile(99),
+        }
+
+    def merge(self, other: "LatencyDigest") -> "LatencyDigest":
+        """Fold ``other`` into this digest (for cross-shard aggregation)."""
+        # pylint: disable=protected-access  # reason: same-class instance merge
+        for bucket, count in other._counts.items():
+            self._counts[bucket] = self._counts.get(bucket, 0) + count
+        self._count += other._count
+        self._sum += other._sum
+        for value in (other._min, other._max):
+            if value is None:
+                continue
+            self._min = value if self._min is None else min(self._min, value)
+            self._max = value if self._max is None else max(self._max, value)
+        return self