Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

## Table of Contents

- [What's new (2026-06-22) — Idempotency-Key Store](#whats-new-2026-06-22--idempotency-key-store)
- [What's new (2026-06-22) — Moving-Average Smoothing](#whats-new-2026-06-22--moving-average-smoothing)
- [What's new (2026-06-22) — Single-Series Anomaly Detection](#whats-new-2026-06-22--single-series-anomaly-detection)
- [What's new (2026-06-22) — Near-Duplicate Text Detection (SimHash / MinHash)](#whats-new-2026-06-22--near-duplicate-text-detection-simhash--minhash)
Expand Down Expand Up @@ -155,6 +156,12 @@

---

## What's new (2026-06-22) — Idempotency-Key Store

Run a side effect once, replay its response on retries. Full reference: [`docs/source/Eng/doc/new_features/v103_features_doc.rst`](docs/source/Eng/doc/new_features/v103_features_doc.rst).

- **`IdempotencyStore` / `request_fingerprint` / `IdempotencyConflict`** (`AC_idempotency_begin`, `AC_idempotency_complete`): `RetryPolicy` re-executes and `work_queue` dedups only in-flight refs — nothing cached the first result. This Stripe-style store returns `new`/`in_progress`/`completed` for a key, replays the stored response, raises on a fingerprint conflict, and supports injectable-clock TTL + JSON persistence. Pure-stdlib, deterministic.

## What's new (2026-06-22) — Moving-Average Smoothing

Smooth a noisy value series. Full reference: [`docs/source/Eng/doc/new_features/v102_features_doc.rst`](docs/source/Eng/doc/new_features/v102_features_doc.rst).
Expand Down
7 changes: 7 additions & 0 deletions README/README_zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

## 目录

- [本次更新 (2026-06-22) — 幂等键存储](#本次更新-2026-06-22--幂等键存储)
- [本次更新 (2026-06-22) — 移动平均平滑](#本次更新-2026-06-22--移动平均平滑)
- [本次更新 (2026-06-22) — 单序列异常检测](#本次更新-2026-06-22--单序列异常检测)
- [本次更新 (2026-06-22) — 近似重复文本检测(SimHash / MinHash)](#本次更新-2026-06-22--近似重复文本检测simhash--minhash)
Expand Down Expand Up @@ -158,6 +159,12 @@

平滑噪声值序列。完整参考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。

## 本次更新 (2026-06-22) — 幂等键存储

副作用只执行一次,重试时重播其响应。完整参考:[`docs/source/Zh/doc/new_features/v103_features_doc.rst`](../docs/source/Zh/doc/new_features/v103_features_doc.rst)。

- **`IdempotencyStore` / `request_fingerprint` / `IdempotencyConflict`**(`AC_idempotency_begin`、`AC_idempotency_complete`):`RetryPolicy` 重试会重跑,`work_queue` 只对进行中引用去重 —— 没有东西缓存第一次结果。本 Stripe 风格存储为某键返回 `new`/`in_progress`/`completed`、重播已存储响应、指纹冲突时抛出异常,并支持可注入时钟 TTL + JSON 持久化。纯标准库、确定。

- **`sma` / `wma` / `ewma` / `rolling`**(`AC_sma`、`AC_ewma`):`stats.describe` 汇总整个样本,`timeseries` 把计数器滚成速率,但没有东西能平滑噪声信号。本功能加入尾端简单/加权/指数加权移动平均与通用滚动归约器,全部返回与输入时间线对齐的等长 list。纯标准库、确定。

## 本次更新 (2026-06-22) — 单序列异常检测
Expand Down
7 changes: 7 additions & 0 deletions README/README_zh-TW.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

## 目錄

- [本次更新 (2026-06-22) — 冪等鍵儲存](#本次更新-2026-06-22--冪等鍵儲存)
- [本次更新 (2026-06-22) — 移動平均平滑](#本次更新-2026-06-22--移動平均平滑)
- [本次更新 (2026-06-22) — 單序列異常偵測](#本次更新-2026-06-22--單序列異常偵測)
- [本次更新 (2026-06-22) — 近似重複文字偵測(SimHash / MinHash)](#本次更新-2026-06-22--近似重複文字偵測simhash--minhash)
Expand Down Expand Up @@ -158,6 +159,12 @@

平滑雜訊值序列。完整參考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。

## 本次更新 (2026-06-22) — 冪等鍵儲存

副作用只執行一次,重試時重播其回應。完整參考:[`docs/source/Zh/doc/new_features/v103_features_doc.rst`](../docs/source/Zh/doc/new_features/v103_features_doc.rst)。

- **`IdempotencyStore` / `request_fingerprint` / `IdempotencyConflict`**(`AC_idempotency_begin`、`AC_idempotency_complete`):`RetryPolicy` 重試會重跑,`work_queue` 只對進行中參照去重 —— 沒有東西快取第一次結果。本 Stripe 風格儲存為某鍵回傳 `new`/`in_progress`/`completed`、重播已儲存回應、指紋衝突時拋出例外,並支援可注入時鐘 TTL + JSON 持久化。純標準函式庫、具決定性。

- **`sma` / `wma` / `ewma` / `rolling`**(`AC_sma`、`AC_ewma`):`stats.describe` 彙總整個樣本,`timeseries` 把計數器滾成速率,但沒有東西能平滑雜訊訊號。本功能加入尾端簡單/加權/指數加權移動平均與通用滾動歸約器,全部回傳與輸入時間線對齊的等長 list。純標準函式庫、具決定性。

## 本次更新 (2026-06-22) — 單序列異常偵測
Expand Down
43 changes: 43 additions & 0 deletions docs/source/Eng/doc/new_features/v103_features_doc.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
Idempotency-Key Store
=====================

``resilience.RetryPolicy`` *re-executes* on retry and ``work_queue`` dedups only
in-flight references — nothing cached the first result so a duplicate request
returns the *same* response without re-running the side effect. This is the
Stripe idempotency pattern: register a key, run the work once, and replay the
stored response for any duplicate.

Pure standard library (``hashlib`` / ``json``); imports no ``PySide6``. The
clock is injectable and the store is in-memory with JSON persistence, so TTL
expiry and replay are fully deterministic in CI.

Headless API
------------

.. code-block:: python

from je_auto_control import IdempotencyStore, request_fingerprint

store = IdempotencyStore(ttl=86400)
state = store.begin("order-42", request_fingerprint(payload))
if state["status"] == "completed":
return state["response"] # replay — do not re-run
result = charge(payload) # run the side effect once
store.complete("order-42", result)

``begin`` returns ``{status, response}`` where status is ``new`` (first time),
``in_progress`` (a duplicate before completion), or ``completed`` (replay the
stored response); reusing a key with a different ``request`` fingerprint raises
``IdempotencyConflict`` (Stripe's HTTP-400 behaviour). ``complete`` records the
response, ``get`` reads a live record, and ``save`` / ``load`` persist the store
as JSON. ``request_fingerprint`` is a stable, order-independent SHA-256 of a
payload.

Executor commands
-----------------

``AC_idempotency_begin`` registers/looks up a ``key`` in a named store (optional
``request`` for conflict detection); ``AC_idempotency_complete`` stores the
``response``. Both use a named-instance registry (like circuit breakers /
bulkheads) and are exposed as MCP tools (``ac_idempotency_begin`` /
``ac_idempotency_complete``) and as Script Builder commands under **Flow**.
1 change: 1 addition & 0 deletions docs/source/Eng/eng_index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ Comprehensive guides for all AutoControl features.
doc/new_features/v100_features_doc
doc/new_features/v101_features_doc
doc/new_features/v102_features_doc
doc/new_features/v103_features_doc
doc/ocr_backends/ocr_backends_doc
doc/observability/observability_doc
doc/operations_layer/operations_layer_doc
Expand Down
35 changes: 35 additions & 0 deletions docs/source/Zh/doc/new_features/v103_features_doc.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
冪等鍵儲存
========

``resilience.RetryPolicy`` 在重試時*重新執行*,``work_queue`` 只對進行中的參照去重 —— 沒有東西快取
第一次的結果,讓重複請求能回傳*相同*回應而不重跑副作用。這是 Stripe 的冪等模式:註冊一個鍵、把工作
執行一次,並為任何重複請求重播已儲存的回應。

純標準函式庫(``hashlib`` / ``json``);不匯入 ``PySide6``。時鐘可注入,儲存為記憶體內並具 JSON 持久化,
因此 TTL 過期與重播在 CI 中完全具決定性。

無頭 API
--------

.. code-block:: python

from je_auto_control import IdempotencyStore, request_fingerprint

store = IdempotencyStore(ttl=86400)
state = store.begin("order-42", request_fingerprint(payload))
if state["status"] == "completed":
return state["response"] # 重播 —— 不要重跑
result = charge(payload) # 副作用只執行一次
store.complete("order-42", result)

``begin`` 回傳 ``{status, response}``,status 為 ``new``(首次)、``in_progress``(完成前的重複)或
``completed``(重播已儲存回應);以不同 ``request`` 指紋重用同一鍵會拋出 ``IdempotencyConflict``
(Stripe 的 HTTP-400 行為)。``complete`` 記錄回應,``get`` 讀取有效記錄,``save`` / ``load`` 以 JSON
持久化。``request_fingerprint`` 是 payload 的穩定、與順序無關的 SHA-256。

執行器命令
----------

``AC_idempotency_begin`` 在具名儲存中註冊/查找 ``key``(可選 ``request`` 做衝突偵測);
``AC_idempotency_complete`` 儲存 ``response``。兩者使用具名實例登錄(如斷路器/隔艙),並以 MCP 工具
(``ac_idempotency_begin`` / ``ac_idempotency_complete``)以及 Script Builder 中 **Flow** 分類下的命令提供。
1 change: 1 addition & 0 deletions docs/source/Zh/zh_index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ AutoControl 所有功能的完整使用指南。
doc/new_features/v100_features_doc
doc/new_features/v101_features_doc
doc/new_features/v102_features_doc
doc/new_features/v103_features_doc
doc/ocr_backends/ocr_backends_doc
doc/observability/observability_doc
doc/operations_layer/operations_layer_doc
Expand Down
5 changes: 5 additions & 0 deletions je_auto_control/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,10 @@
from je_auto_control.utils.resilience import (
CircuitBreaker, CircuitOpenError, RetryPolicy, retry_call,
)
# Idempotency-key store with stored responses (Stripe-style request dedup)
from je_auto_control.utils.idempotency import (
IdempotencyConflict, IdempotencyStore, request_fingerprint,
)
# CI workflow annotations (GitHub Actions)
from je_auto_control.utils.ci_annotations import (
emit_annotations, format_annotation,
Expand Down Expand Up @@ -925,6 +929,7 @@ def start_autocontrol_gui(*args, **kwargs):
"snapshot_screen",
"replay_timeline", "run_sequence",
"CircuitBreaker", "CircuitOpenError", "RetryPolicy", "retry_call",
"IdempotencyConflict", "IdempotencyStore", "request_fingerprint",
"emit_annotations", "format_annotation",
"ClipboardHistory", "default_clipboard_history",
"analyze_heal_log", "heal_stats", "scan_secrets",
Expand Down
19 changes: 19 additions & 0 deletions je_auto_control/gui/script_builder/command_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -1995,6 +1995,25 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
),
description="Exponentially-weighted moving average of a series.",
))
specs.append(CommandSpec(
"AC_idempotency_begin", "Flow", "Idempotency: Begin",
fields=(
FieldSpec("name", FieldType.STRING, placeholder="payments"),
FieldSpec("key", FieldType.STRING, placeholder="order-42"),
FieldSpec("request", FieldType.STRING, optional=True,
placeholder='{"amount": 100}'),
),
description="Register/look up an idempotency key (new/in_progress/done).",
))
specs.append(CommandSpec(
"AC_idempotency_complete", "Flow", "Idempotency: Complete",
fields=(
FieldSpec("name", FieldType.STRING, placeholder="payments"),
FieldSpec("key", FieldType.STRING, placeholder="order-42"),
FieldSpec("response", FieldType.STRING, placeholder='{"ok": true}'),
),
description="Store the completed response for an idempotency key.",
))
specs.append(CommandSpec(
"AC_diff_rows", "Data", "Dataset Diff: Rows by Key",
fields=(
Expand Down
22 changes: 22 additions & 0 deletions je_auto_control/utils/executor/action_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -2929,6 +2929,26 @@ def _rate_limit(name: str, rate: float = 1.0, capacity: float = 1.0,


_BULKHEADS: Dict[str, Any] = {}
_IDEMPOTENCY_STORES: Dict[str, Any] = {}


def _idempotency_begin(name: str, key: str,
request: Any = None) -> Dict[str, Any]:
"""Adapter: register/look up an idempotency key in a named store."""
from je_auto_control.utils.idempotency import (
IdempotencyStore, request_fingerprint)
store = _IDEMPOTENCY_STORES.setdefault(name, IdempotencyStore())
fingerprint = request_fingerprint(request) if request is not None else None
return store.begin(key, fingerprint)


def _idempotency_complete(name: str, key: str,
response: Any) -> Dict[str, Any]:
"""Adapter: store the completed response for an idempotency key."""
from je_auto_control.utils.idempotency import IdempotencyStore
store = _IDEMPOTENCY_STORES.setdefault(name, IdempotencyStore())
store.complete(key, response)
return {"status": "completed"}


def _bulkhead_run(name: str, max_concurrent: int,
Expand Down Expand Up @@ -4544,6 +4564,8 @@ def __init__(self):
"AC_detect_anomalies": _detect_anomalies,
"AC_sma": _sma,
"AC_ewma": _ewma,
"AC_idempotency_begin": _idempotency_begin,
"AC_idempotency_complete": _idempotency_complete,
"AC_detect_drift": _detect_drift,
"AC_categorical_drift": _categorical_drift,
"AC_diff_rows": _diff_rows,
Expand Down
6 changes: 6 additions & 0 deletions je_auto_control/utils/idempotency/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
"""Idempotency-key store with stored responses for AutoControl."""
from je_auto_control.utils.idempotency.idempotency import (
IdempotencyConflict, IdempotencyStore, request_fingerprint,
)

__all__ = ["IdempotencyConflict", "IdempotencyStore", "request_fingerprint"]
108 changes: 108 additions & 0 deletions je_auto_control/utils/idempotency/idempotency.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""Idempotency-key store with stored responses (Stripe-style request dedup).

``resilience.RetryPolicy`` *re-executes* on retry, and ``work_queue`` dedups
only in-flight references — nothing caches the first result so a duplicate
request returns the *same* response without re-running the side effect. This is
the missing half of the retry story: register a key, run the work once, and
replay the stored response for any duplicate.

Pure standard library (``hashlib`` / ``json``); imports no ``PySide6``. The
clock is injectable and the store is in-memory (with JSON persistence), so TTL
expiry and replay are fully deterministic in CI.
"""
import hashlib
import json
import time
from pathlib import Path
from typing import Any, Callable, Dict, Optional

from je_auto_control.utils.exception.exceptions import AutoControlException


class IdempotencyConflict(AutoControlException):
"""A key was reused with a different request fingerprint."""


def request_fingerprint(payload: Any) -> str:
"""Return a stable SHA-256 fingerprint of a request payload."""
encoded = json.dumps(payload, sort_keys=True, default=str).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()


class IdempotencyStore:
"""Stores per-key request state and the first completed response."""

def __init__(self, *, clock: Callable[[], float] = time.time,
ttl: Optional[float] = None) -> None:
self._records: Dict[str, Dict[str, Any]] = {}
self._clock = clock
self._ttl = ttl

def _live(self, key: str) -> Optional[Dict[str, Any]]:
record = self._records.get(key)
if record is None:
return None
if self._ttl is not None and self._clock() - record["stored_at"] >= self._ttl:
self._records.pop(key, None)
return None
return record

def begin(self, key: str,
request: Optional[str] = None) -> Dict[str, Any]:
"""Register ``key`` or return its existing state.

Returns ``{status, response}`` where status is ``new`` (first time),
``in_progress`` (a duplicate before completion), or ``completed`` (replay
the stored response). Raises :class:`IdempotencyConflict` when ``key`` is
reused with a different ``request`` fingerprint.
"""
record = self._live(key)
if record is not None:
if (request is not None and record["request"] is not None
and record["request"] != request):
raise IdempotencyConflict(
f"key {key!r} reused with a different request")
return {"status": record["status"], "response": record["response"]}
self._records[key] = {"status": "in_progress", "request": request,
"response": None, "stored_at": self._clock()}
return {"status": "new", "response": None}

def complete(self, key: str, response: Any) -> None:
"""Record the completed ``response`` for ``key``."""
record = self._records.get(key)
if record is None:
self._records[key] = {"status": "completed", "request": None,
"response": response,
"stored_at": self._clock()}
else:
record["status"] = "completed"
record["response"] = response

def get(self, key: str) -> Optional[Dict[str, Any]]:
"""Return the live record for ``key`` (or ``None`` if absent/expired)."""
record = self._live(key)
return dict(record) if record is not None else None

def to_dict(self) -> Dict[str, Any]:
"""Return all records as a plain dict."""
return {key: dict(value) for key, value in self._records.items()}

@classmethod
def from_dict(cls, data: Dict[str, Any], **kwargs: Any) -> "IdempotencyStore":
"""Build a store from a :meth:`to_dict` mapping."""
store = cls(**kwargs)
store._records = {key: dict(value) for key, value in data.items()}
return store

def save(self, path: str) -> str:
"""Persist the store to ``path`` as JSON; return the path."""
out = Path(path)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(self.to_dict(), indent=2), encoding="utf-8")
return str(out)

@classmethod
def load(cls, path: str, **kwargs: Any) -> "IdempotencyStore":
"""Load a store from a JSON file."""
data = json.loads(Path(path).read_text(encoding="utf-8"))
return cls.from_dict(data, **kwargs)
30 changes: 29 additions & 1 deletion je_auto_control/utils/mcp_server/tools/_factories.py
Original file line number Diff line number Diff line change
Expand Up @@ -3548,6 +3548,34 @@ def dataset_diff_tools() -> List[MCPTool]:
]


def idempotency_tools() -> List[MCPTool]:
return [
MCPTool(
name="ac_idempotency_begin",
description=("Register/look up idempotency 'key' in a named store "
"'name' (optional 'request' for conflict detection). "
"Returns {status: new|in_progress|completed, response}."),
input_schema=schema(
{"name": {"type": "string"}, "key": {"type": "string"},
"request": {"type": "object"}},
["name", "key"]),
handler=h.idempotency_begin,
annotations=READ_ONLY,
),
MCPTool(
name="ac_idempotency_complete",
description=("Store the completed 'response' for idempotency 'key' "
"in named store 'name'. Returns {status}."),
input_schema=schema(
{"name": {"type": "string"}, "key": {"type": "string"},
"response": {"type": "object"}},
["name", "key", "response"]),
handler=h.idempotency_complete,
annotations=NON_DESTRUCTIVE,
),
]


def smoothing_tools() -> List[MCPTool]:
return [
MCPTool(
Expand Down Expand Up @@ -5524,7 +5552,7 @@ def media_assert_tools() -> List[MCPTool]:
secret_ref_tools, config_schema_tools, config_redaction_tools,
data_profile_tools, http_problem_tools, dotenv_tools,
sse_client_tools, layered_config_tools, data_drift_tools, schema_compat_tools,
timeseries_tools, anomaly_tools, smoothing_tools,
timeseries_tools, anomaly_tools, smoothing_tools, idempotency_tools,
dataset_diff_tools, referential_tools, link_header_tools, multipart_tools,
http_content_tools, cookie_jar_tools, http_conditional_tools,
saga_tools, decision_table_tools, locator_repair_tools,
Expand Down
Loading
Loading