Merge pull request #314 from Integration-Automation/feat/optimistic-batch

JE-Chen · web-flow · commit dbc24a76e34a · 2026-06-22T10:42:22.000+08:00
Add optimistic-concurrency versioned store (CAS / If-Match)
diff --git a/README.md b/README.md
@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-22) — Optimistic-Concurrency Versioned Store](#whats-new-2026-06-22--optimistic-concurrency-versioned-store)
 - [What's new (2026-06-22) — Per-Stream Sequence-Gap Detection](#whats-new-2026-06-22--per-stream-sequence-gap-detection)
 - [What's new (2026-06-22) — Time-Windowed Deduplication](#whats-new-2026-06-22--time-windowed-deduplication)
 - [What's new (2026-06-22) — Idempotency-Key Store](#whats-new-2026-06-22--idempotency-key-store)
@@ -158,6 +159,12 @@
 
 ---
 
+## What's new (2026-06-22) — Optimistic-Concurrency Versioned Store
+
+Update only if the version is unchanged (compare-and-swap / If-Match). Full reference: [`docs/source/Eng/doc/new_features/v106_features_doc.rst`](docs/source/Eng/doc/new_features/v106_features_doc.rst).
+
+- **`VersionedStore` / `VersionConflict` / `if_match_header` / `check_if_match`** (`AC_cas_put`, `AC_cas_get`): `http_conditional` used ETag for read caching but never for write concurrency. This local compare-and-swap store `put`s only when `expected_version` matches (raising `VersionConflict` on a stale write), bumps a monotonic version, and bridges to HTTP `If-Match` — the write side of the ETag story. Pure-stdlib, deterministic.
+
 ## What's new (2026-06-22) — Per-Stream Sequence-Gap Detection
 
 Detect missing / out-of-order / duplicate messages by sequence number. Full reference: [`docs/source/Eng/doc/new_features/v105_features_doc.rst`](docs/source/Eng/doc/new_features/v105_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) — 幂等键存储](#本次更新-2026-06-22--幂等键存储)
@@ -161,6 +162,12 @@
 
 平滑噪声值序列。完整参考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。
 
+## 本次更新 (2026-06-22) — 乐观并发版本存储
+
+只在版本未变时更新(compare-and-swap / If-Match)。完整参考:[`docs/source/Zh/doc/new_features/v106_features_doc.rst`](../docs/source/Zh/doc/new_features/v106_features_doc.rst)。
+
+- **`VersionedStore` / `VersionConflict` / `if_match_header` / `check_if_match`**(`AC_cas_put`、`AC_cas_get`):`http_conditional` 以 ETag 做读取缓存,但从不用于写入并发。本地 compare-and-swap 存储仅在 `expected_version` 相符时 `put`(过时写入抛出 `VersionConflict`)、递增单调版本,并桥接到 HTTP `If-Match` —— ETag 故事的写入面。纯标准库、确定。
+
 ## 本次更新 (2026-06-22) — 逐流序号间隙检测
 
 按序号检测遗漏/乱序/重复的消息。完整参考:[`docs/source/Zh/doc/new_features/v105_features_doc.rst`](../docs/source/Zh/doc/new_features/v105_features_doc.rst)。
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) — 冪等鍵儲存](#本次更新-2026-06-22--冪等鍵儲存)
@@ -161,6 +162,12 @@
 
 平滑雜訊值序列。完整參考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。
 
+## 本次更新 (2026-06-22) — 樂觀並行版本儲存
+
+只在版本未變時更新(compare-and-swap / If-Match)。完整參考:[`docs/source/Zh/doc/new_features/v106_features_doc.rst`](../docs/source/Zh/doc/new_features/v106_features_doc.rst)。
+
+- **`VersionedStore` / `VersionConflict` / `if_match_header` / `check_if_match`**(`AC_cas_put`、`AC_cas_get`):`http_conditional` 以 ETag 做讀取快取,但從不用於寫入並行。本地 compare-and-swap 儲存僅在 `expected_version` 相符時 `put`(過時寫入拋出 `VersionConflict`)、遞增單調版本,並橋接到 HTTP `If-Match` —— ETag 故事的寫入面。純標準函式庫、具決定性。
+
 ## 本次更新 (2026-06-22) — 逐串流序號間隙偵測
 
 依序號偵測遺漏/亂序/重複的訊息。完整參考:[`docs/source/Zh/doc/new_features/v105_features_doc.rst`](../docs/source/Zh/doc/new_features/v105_features_doc.rst)。
diff --git a/docs/source/Eng/doc/new_features/v106_features_doc.rst b/docs/source/Eng/doc/new_features/v106_features_doc.rst
@@ -0,0 +1,42 @@
+Optimistic-Concurrency Versioned Store
+======================================
+
+``http_conditional`` uses ETag for *read* caching (``If-None-Match`` / 304) but
+never for *write* concurrency (``If-Match`` / version check). There was no local
+compare-and-swap / versioned record store for "update only if the version is
+unchanged". This fills the write side of the ETag story.
+
+Pure standard library (``json``); imports no ``PySide6``. The version is a
+monotonic int and the store is in-memory with JSON persistence, so behaviour is
+fully deterministic in CI.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import VersionedStore, VersionConflict, if_match_header
+
+    store = VersionedStore()
+    version = store.put("db.host", "prod-1")          # version 1
+    record = store.get("db.host")                      # {"value": ..., "version": 1}
+    try:
+        store.put("db.host", "prod-2", expected_version=record["version"])
+    except VersionConflict:
+        reload_and_retry()
+    header = if_match_header(version)                  # '"1"' for an HTTP If-Match
+
+``put`` writes only when ``expected_version`` matches the current version
+(``0`` requires the key to be absent, omitting it is a blind write) and returns
+the new version, raising ``VersionConflict`` on a stale write. ``get`` returns
+``{value, version}``; ``delete`` is likewise guarded; ``save`` / ``load``
+persist as JSON. ``if_match_header`` / ``check_if_match`` bridge to real HTTP
+``If-Match`` writes alongside ``http_conditional``.
+
+Executor commands
+-----------------
+
+``AC_cas_put`` returns ``{ok, version}`` (or ``{ok: false, error}`` on conflict);
+``AC_cas_get`` returns ``{record}``. Both use a named-instance registry and are
+exposed as MCP tools (``ac_cas_put`` / ``ac_cas_get``) and as Script Builder
+commands under **Flow**.
diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst
@@ -128,6 +128,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v103_features_doc
    doc/new_features/v104_features_doc
    doc/new_features/v105_features_doc
+   doc/new_features/v106_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/v106_features_doc.rst b/docs/source/Zh/doc/new_features/v106_features_doc.rst
@@ -0,0 +1,37 @@
+樂觀並行版本儲存
+==============
+
+``http_conditional`` 以 ETag 做*讀取*快取(``If-None-Match`` / 304),但從不用於*寫入*並行
+(``If-Match`` / 版本檢查)。沒有本地的 compare-and-swap / 版本化記錄儲存來做「只在版本未變時更新」。
+本功能補上 ETag 故事的寫入面。
+
+純標準函式庫(``json``);不匯入 ``PySide6``。版本為單調整數,儲存為記憶體內並具 JSON 持久化,因此行為
+在 CI 中完全具決定性。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import VersionedStore, VersionConflict, if_match_header
+
+    store = VersionedStore()
+    version = store.put("db.host", "prod-1")          # 版本 1
+    record = store.get("db.host")                      # {"value": ..., "version": 1}
+    try:
+        store.put("db.host", "prod-2", expected_version=record["version"])
+    except VersionConflict:
+        reload_and_retry()
+    header = if_match_header(version)                  # HTTP If-Match 用 '"1"'
+
+``put`` 僅在 ``expected_version`` 與當前版本相符時寫入(``0`` 要求鍵不存在,省略則為盲寫)並回傳新版本,
+過時寫入時拋出 ``VersionConflict``。``get`` 回傳 ``{value, version}``;``delete`` 同樣受保護;``save`` /
+``load`` 以 JSON 持久化。``if_match_header`` / ``check_if_match`` 與 ``http_conditional`` 搭配,橋接到真正的
+HTTP ``If-Match`` 寫入。
+
+執行器命令
+----------
+
+``AC_cas_put`` 回傳 ``{ok, version}``(衝突時 ``{ok: false, error}``);``AC_cas_get`` 回傳 ``{record}``。
+兩者使用具名實例登錄,並以 MCP 工具(``ac_cas_put`` / ``ac_cas_get``)以及 Script Builder 中 **Flow** 分類下
+的命令提供。
diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst
@@ -128,6 +128,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v103_features_doc
    doc/new_features/v104_features_doc
    doc/new_features/v105_features_doc
+   doc/new_features/v106_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
@@ -207,6 +207,10 @@
 from je_auto_control.utils.dedup_window import DedupWindow
 # Per-stream sequence-gap / ordering detection
 from je_auto_control.utils.sequence_gap import SequenceTracker
+# Optimistic-concurrency versioned store (compare-and-swap / If-Match)
+from je_auto_control.utils.optimistic import (
+    VersionConflict, VersionedStore, check_if_match, if_match_header,
+)
 # CI workflow annotations (GitHub Actions)
 from je_auto_control.utils.ci_annotations import (
     emit_annotations, format_annotation,
@@ -935,6 +939,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "CircuitBreaker", "CircuitOpenError", "RetryPolicy", "retry_call",
     "IdempotencyConflict", "IdempotencyStore", "request_fingerprint",
     "DedupWindow", "SequenceTracker",
+    "VersionConflict", "VersionedStore", "check_if_match", "if_match_header",
     "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
@@ -2032,6 +2032,24 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Classify a sequence number (ok/duplicate/gap/reorder).",
     ))
+    specs.append(CommandSpec(
+        "AC_cas_put", "Flow", "Optimistic: Put (CAS)",
+        fields=(
+            FieldSpec("name", FieldType.STRING, placeholder="config"),
+            FieldSpec("key", FieldType.STRING, placeholder="db.host"),
+            FieldSpec("value", FieldType.STRING, placeholder='"prod-1"'),
+            FieldSpec("expected_version", FieldType.INT, optional=True),
+        ),
+        description="Put only if expected_version matches (returns new version).",
+    ))
+    specs.append(CommandSpec(
+        "AC_cas_get", "Flow", "Optimistic: Get",
+        fields=(
+            FieldSpec("name", FieldType.STRING, placeholder="config"),
+            FieldSpec("key", FieldType.STRING, placeholder="db.host"),
+        ),
+        description="Read a versioned record {value, version}.",
+    ))
     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
@@ -2932,6 +2932,33 @@ def _rate_limit(name: str, rate: float = 1.0, capacity: float = 1.0,
 _IDEMPOTENCY_STORES: Dict[str, Any] = {}
 _DEDUP_WINDOWS: Dict[str, Any] = {}
 _SEQUENCE_TRACKERS: Dict[str, Any] = {}
+_VERSIONED_STORES: Dict[str, Any] = {}
+
+
+def _cas_put(name: str, key: str, value: Any,
+             expected_version: Any = None) -> Dict[str, Any]:
+    """Adapter: optimistic put into a named versioned store."""
+    import json
+    from je_auto_control.utils.optimistic import VersionConflict, VersionedStore
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except ValueError:
+            pass
+    store = _VERSIONED_STORES.setdefault(name, VersionedStore())
+    expected = int(expected_version) if expected_version is not None else None
+    try:
+        version = store.put(key, value, expected_version=expected)
+    except VersionConflict as error:
+        return {"ok": False, "error": str(error)}
+    return {"ok": True, "version": version}
+
+
+def _cas_get(name: str, key: str) -> Dict[str, Any]:
+    """Adapter: read a record from a named versioned store."""
+    from je_auto_control.utils.optimistic import VersionedStore
+    store = _VERSIONED_STORES.setdefault(name, VersionedStore())
+    return {"record": store.get(key)}
 
 
 def _sequence_observe(name: str, stream_id: str, seq: Any) -> Dict[str, Any]:
@@ -4586,6 +4613,8 @@ def __init__(self):
             "AC_idempotency_complete": _idempotency_complete,
             "AC_dedup_check": _dedup_check,
             "AC_sequence_observe": _sequence_observe,
+            "AC_cas_put": _cas_put,
+            "AC_cas_get": _cas_get,
             "AC_detect_drift": _detect_drift,
             "AC_categorical_drift": _categorical_drift,
             "AC_diff_rows": _diff_rows,
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 optimistic_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_cas_put",
+            description=("Optimistic put 'value' at 'key' in named store 'name' "
+                         "if 'expected_version' matches (0=absent, omit=blind). "
+                         "Returns {ok, version} or {ok: false, error}."),
+            input_schema=schema(
+                {"name": {"type": "string"}, "key": {"type": "string"},
+                 "value": {"type": "object"},
+                 "expected_version": {"type": "integer"}},
+                ["name", "key", "value"]),
+            handler=h.cas_put,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_cas_get",
+            description=("Read {record: {value, version}} (or null) for 'key' in "
+                         "named versioned store 'name'."),
+            input_schema=schema(
+                {"name": {"type": "string"}, "key": {"type": "string"}},
+                ["name", "key"]),
+            handler=h.cas_get,
+            annotations=READ_ONLY,
+        ),
+    ]
+
+
 def sequence_gap_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -5587,7 +5615,7 @@ def media_assert_tools() -> List[MCPTool]:
     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, idempotency_tools,
-    dedup_window_tools, sequence_gap_tools,
+    dedup_window_tools, sequence_gap_tools, optimistic_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
@@ -1942,6 +1942,16 @@ def sequence_observe(name, stream_id, seq):
     return _sequence_observe(name, stream_id, seq)
 
 
+def cas_put(name, key, value, expected_version=None):
+    from je_auto_control.utils.executor.action_executor import _cas_put
+    return _cas_put(name, key, value, expected_version)
+
+
+def cas_get(name, key):
+    from je_auto_control.utils.executor.action_executor import _cas_get
+    return _cas_get(name, key)
+
+
 def detect_drift(reference, current, threshold=0.25, bins=10):
     from je_auto_control.utils.executor.action_executor import _detect_drift
     return _detect_drift(reference, current, threshold, bins)
diff --git a/je_auto_control/utils/optimistic/__init__.py b/je_auto_control/utils/optimistic/__init__.py
@@ -0,0 +1,8 @@
+"""Optimistic-concurrency versioned store for AutoControl."""
+from je_auto_control.utils.optimistic.optimistic import (
+    VersionConflict, VersionedStore, check_if_match, if_match_header,
+)
+
+__all__ = [
+    "VersionConflict", "VersionedStore", "check_if_match", "if_match_header",
+]
diff --git a/je_auto_control/utils/optimistic/optimistic.py b/je_auto_control/utils/optimistic/optimistic.py
diff --git a/test/unit_test/headless/test_optimistic_batch.py b/test/unit_test/headless/test_optimistic_batch.py
