Add idempotency-key store with stored responses

JE-Chen · JE-Chen · commit 9db1b0806ab6 · 2026-06-22T10:05:03.000+08:00
RetryPolicy re-executes on retry and work_queue dedups only in-flight
references — nothing cached the first result so a duplicate returned the
same response without re-running the side effect. Add a Stripe-style
IdempotencyStore (begin → new/in_progress/completed, complete, replay,
fingerprint-conflict detection, injectable-clock TTL, JSON persistence)
and request_fingerprint. Wired through facade, executor (AC_idempotency_
begin / AC_idempotency_complete, named-instance registry), MCP, and the
Script Builder with a headless test batch and EN/Zh docs.
diff --git a/README.md b/README.md
@@ -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)
@@ -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).
diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md
@@ -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)
@@ -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) — 单序列异常检测
diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md
@@ -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)
@@ -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) — 單序列異常偵測
diff --git a/docs/source/Eng/doc/new_features/v103_features_doc.rst b/docs/source/Eng/doc/new_features/v103_features_doc.rst
@@ -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**.
diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst
@@ -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
diff --git a/docs/source/Zh/doc/new_features/v103_features_doc.rst b/docs/source/Zh/doc/new_features/v103_features_doc.rst
@@ -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** 分類下的命令提供。
diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst
@@ -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
diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py
@@ -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,
@@ -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",
diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py
@@ -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=(
diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py
@@ -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,
@@ -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,
diff --git a/je_auto_control/utils/idempotency/__init__.py b/je_auto_control/utils/idempotency/__init__.py
@@ -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"]
diff --git a/je_auto_control/utils/idempotency/idempotency.py b/je_auto_control/utils/idempotency/idempotency.py
@@ -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)
diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py
@@ -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(
@@ -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,
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_idempotency_batch.py b/test/unit_test/headless/test_idempotency_batch.py
