From 2bc97ad9d39350a0de5338caf52a6e98a589ffdd Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Mon, 22 Jun 2026 10:15:22 +0800 Subject: [PATCH] Add time-windowed message deduplication (exactly-once inbox) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit work_queue dedups only new/in_progress references — once an item completes the same reference enqueues again and redelivered webhooks reprocess. Add DedupWindow: a sliding-window inbox whose check_and_mark returns True the first time an id is seen within ttl_s and False for a duplicate, converting at-least-once delivery to exactly-once-in-window. Injectable clock, bounded size. Wired through facade, executor (AC_dedup_check, named-instance registry), MCP, and the Script Builder with a headless test batch and EN/Zh docs. --- README.md | 7 ++ README/README_zh-CN.md | 7 ++ README/README_zh-TW.md | 7 ++ .../doc/new_features/v104_features_doc.rst | 37 +++++++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v104_features_doc.rst | 32 ++++++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 3 + .../gui/script_builder/command_schema.py | 9 +++ .../utils/dedup_window/__init__.py | 4 ++ .../utils/dedup_window/dedup_window.py | 56 ++++++++++++++++ .../utils/executor/action_executor.py | 11 ++++ .../utils/mcp_server/tools/_factories.py | 18 ++++++ .../utils/mcp_server/tools/_handlers.py | 5 ++ .../headless/test_dedup_window_batch.py | 64 +++++++++++++++++++ 15 files changed, 262 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v104_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v104_features_doc.rst create mode 100644 je_auto_control/utils/dedup_window/__init__.py create mode 100644 je_auto_control/utils/dedup_window/dedup_window.py create mode 100644 test/unit_test/headless/test_dedup_window_batch.py diff --git a/README.md b/README.md index 6d97ecdb..5faa668e 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [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) - [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) @@ -156,6 +157,12 @@ --- +## What's new (2026-06-22) — Time-Windowed Deduplication + +Drop duplicate/redelivered messages within a TTL window. Full reference: [`docs/source/Eng/doc/new_features/v104_features_doc.rst`](docs/source/Eng/doc/new_features/v104_features_doc.rst). + +- **`DedupWindow`** (`AC_dedup_check`): `work_queue` dedups only in-flight references, so a completed reference re-enqueues and redelivered webhooks reprocess. This sliding-window inbox `check_and_mark`s a message id — `True` the first time, `False` for a duplicate within `ttl_s` — converting at-least-once delivery to exactly-once-in-window. Injectable clock, bounded size. Pure-stdlib, deterministic. + ## 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). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index 5072af1c..f91690ee 100644 --- 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--单序列异常检测) @@ -159,6 +160,12 @@ 平滑噪声值序列。完整参考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。 +## 本次更新 (2026-06-22) — 时间窗口去重 + +在 TTL 窗口内丢弃重复/重送的消息。完整参考:[`docs/source/Zh/doc/new_features/v104_features_doc.rst`](../docs/source/Zh/doc/new_features/v104_features_doc.rst)。 + +- **`DedupWindow`**(`AC_dedup_check`):`work_queue` 只对进行中引用去重,因此已完成的引用会重新入列、重送的 webhook 会重复处理。本滑动窗口收件箱对消息 id 做 `check_and_mark` —— 首次返回 `True`、`ttl_s` 窗口内重复返回 `False` —— 把至少一次投递转换成窗口内恰好一次。可注入时钟、大小有界。纯标准库、确定。 + ## 本次更新 (2026-06-22) — 幂等键存储 副作用只执行一次,重试时重播其响应。完整参考:[`docs/source/Zh/doc/new_features/v103_features_doc.rst`](../docs/source/Zh/doc/new_features/v103_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index b2d4bbe2..49ab3ce4 100644 --- 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--單序列異常偵測) @@ -159,6 +160,12 @@ 平滑雜訊值序列。完整參考:[`docs/source/Zh/doc/new_features/v102_features_doc.rst`](../docs/source/Zh/doc/new_features/v102_features_doc.rst)。 +## 本次更新 (2026-06-22) — 時間視窗去重 + +在 TTL 視窗內丟棄重複/重送的訊息。完整參考:[`docs/source/Zh/doc/new_features/v104_features_doc.rst`](../docs/source/Zh/doc/new_features/v104_features_doc.rst)。 + +- **`DedupWindow`**(`AC_dedup_check`):`work_queue` 只對進行中參照去重,因此已完成的參照會重新入列、重送的 webhook 會重複處理。本滑動視窗收件匣對訊息 id 做 `check_and_mark` —— 首次回傳 `True`、`ttl_s` 視窗內重複回傳 `False` —— 把至少一次投遞轉換成視窗內恰好一次。可注入時鐘、大小有界。純標準函式庫、具決定性。 + ## 本次更新 (2026-06-22) — 冪等鍵儲存 副作用只執行一次,重試時重播其回應。完整參考:[`docs/source/Zh/doc/new_features/v103_features_doc.rst`](../docs/source/Zh/doc/new_features/v103_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v104_features_doc.rst b/docs/source/Eng/doc/new_features/v104_features_doc.rst new file mode 100644 index 00000000..4f045025 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v104_features_doc.rst @@ -0,0 +1,37 @@ +Time-Windowed Deduplication +=========================== + +``work_queue`` dedups only ``new`` / ``in_progress`` references — once an item +completes, the same reference enqueues again, and redelivered webhooks are +reprocessed. This adds the missing "seen this id in the last N seconds → drop +it" inbox that converts at-least-once delivery to exactly-once-in-window. + +Pure standard library; imports no ``PySide6``. The clock is injectable, so TTL +eviction is fully deterministic in CI. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import DedupWindow + + inbox = DedupWindow(ttl_s=3600) + if inbox.check_and_mark(event_id): + process(event) # first time within the window + else: + skip(event) # duplicate / redelivery + +``check_and_mark`` atomically returns ``True`` the first time an id is seen +within the window (and marks it) or ``False`` for a duplicate. ``seen`` / ``mark`` +are the separate query/record halves, ``purge_expired`` drops stale entries, and +``size`` reports the live count. Entries older than ``ttl_s`` are evicted on each +operation, so the window stays bounded. + +Executor command +---------------- + +``AC_dedup_check`` check-and-marks a ``message_id`` in a named window (TTL +``ttl_s``) and returns ``{first_seen, size}``. It uses a named-instance registry +and is exposed as the MCP tool ``ac_dedup_check`` and as a Script Builder +command under **Flow**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 35dd8c07..6ba95b4c 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -126,6 +126,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v101_features_doc doc/new_features/v102_features_doc doc/new_features/v103_features_doc + doc/new_features/v104_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/v104_features_doc.rst b/docs/source/Zh/doc/new_features/v104_features_doc.rst new file mode 100644 index 00000000..06f45b4c --- /dev/null +++ b/docs/source/Zh/doc/new_features/v104_features_doc.rst @@ -0,0 +1,32 @@ +時間視窗去重 +========== + +``work_queue`` 只對 ``new`` / ``in_progress`` 參照去重 —— 一旦項目完成,同一參照又會再入列,重送的 +webhook 也會被重複處理。本功能補上缺少的「最近 N 秒看過這個 id → 丟棄」收件匣,把至少一次投遞轉換成 +視窗內恰好一次。 + +純標準函式庫;不匯入 ``PySide6``。時鐘可注入,因此 TTL 驅逐在 CI 中完全具決定性。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import DedupWindow + + inbox = DedupWindow(ttl_s=3600) + if inbox.check_and_mark(event_id): + process(event) # 視窗內首次 + else: + skip(event) # 重複 / 重送 + +``check_and_mark`` 在視窗內首次看到某 id 時原子性回傳 ``True``(並標記),重複則回傳 ``False``。``seen`` / +``mark`` 是分離的查詢/記錄兩半,``purge_expired`` 丟棄過期項目,``size`` 回報有效數量。超過 ``ttl_s`` 的 +項目會在每次操作時驅逐,因此視窗保持有界。 + +執行器命令 +---------- + +``AC_dedup_check`` 在具名視窗(TTL ``ttl_s``)中對 ``message_id`` 做 check-and-mark,回傳 +``{first_seen, size}``。它使用具名實例登錄,並以 MCP 工具 ``ac_dedup_check`` 以及 Script Builder 中 +**Flow** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index a9cc3e86..e0d936f3 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -126,6 +126,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v101_features_doc doc/new_features/v102_features_doc doc/new_features/v103_features_doc + doc/new_features/v104_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 index b3ea0d4c..cb01f5bf 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -203,6 +203,8 @@ from je_auto_control.utils.idempotency import ( IdempotencyConflict, IdempotencyStore, request_fingerprint, ) +# Time-windowed message deduplication (exactly-once inbox) +from je_auto_control.utils.dedup_window import DedupWindow # CI workflow annotations (GitHub Actions) from je_auto_control.utils.ci_annotations import ( emit_annotations, format_annotation, @@ -930,6 +932,7 @@ def start_autocontrol_gui(*args, **kwargs): "replay_timeline", "run_sequence", "CircuitBreaker", "CircuitOpenError", "RetryPolicy", "retry_call", "IdempotencyConflict", "IdempotencyStore", "request_fingerprint", + "DedupWindow", "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 index 81916617..204e41ca 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -2014,6 +2014,15 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None: ), description="Store the completed response for an idempotency key.", )) + specs.append(CommandSpec( + "AC_dedup_check", "Flow", "Dedup Window: Check", + fields=( + FieldSpec("name", FieldType.STRING, placeholder="webhooks"), + FieldSpec("message_id", FieldType.STRING, placeholder="evt-123"), + FieldSpec("ttl_s", FieldType.FLOAT, optional=True, default=3600), + ), + description="Check-and-mark a message id; first_seen false on duplicate.", + )) specs.append(CommandSpec( "AC_diff_rows", "Data", "Dataset Diff: Rows by Key", fields=( diff --git a/je_auto_control/utils/dedup_window/__init__.py b/je_auto_control/utils/dedup_window/__init__.py new file mode 100644 index 00000000..a3742f2a --- /dev/null +++ b/je_auto_control/utils/dedup_window/__init__.py @@ -0,0 +1,4 @@ +"""Time-windowed message deduplication for AutoControl.""" +from je_auto_control.utils.dedup_window.dedup_window import DedupWindow + +__all__ = ["DedupWindow"] diff --git a/je_auto_control/utils/dedup_window/dedup_window.py b/je_auto_control/utils/dedup_window/dedup_window.py new file mode 100644 index 00000000..d9f2ab10 --- /dev/null +++ b/je_auto_control/utils/dedup_window/dedup_window.py @@ -0,0 +1,56 @@ +"""Time-windowed deduplication (at-least-once → exactly-once inbox). + +``work_queue`` dedups only ``new`` / ``in_progress`` references — once an item +completes, the same reference enqueues again, and redelivered webhooks are +reprocessed. This is the missing "seen this id in the last N seconds → drop it" +inbox that converts at-least-once delivery to exactly-once-in-window. + +Pure standard library; imports no ``PySide6``. The clock is injectable, so TTL +eviction is fully deterministic in CI. +""" +import time +from typing import Callable, Dict + + +class DedupWindow: + """A sliding time window of recently-seen message ids.""" + + def __init__(self, ttl_s: float, *, + clock: Callable[[], float] = time.time) -> None: + self._ttl = float(ttl_s) + self._clock = clock + self._seen: Dict[str, float] = {} + + def _purge(self, now: float) -> None: + cutoff = now - self._ttl + expired = [key for key, ts in self._seen.items() if ts <= cutoff] + for key in expired: + del self._seen[key] + + def seen(self, message_id: str) -> bool: + """Whether ``message_id`` is in the window and not expired.""" + now = self._clock() + self._purge(now) + return message_id in self._seen + + def mark(self, message_id: str) -> None: + """Record ``message_id`` as seen now.""" + self._seen[str(message_id)] = self._clock() + + def check_and_mark(self, message_id: str) -> bool: + """Atomically return ``True`` if first-seen (and mark), else ``False``.""" + now = self._clock() + self._purge(now) + if message_id in self._seen: + return False + self._seen[str(message_id)] = now + return True + + def purge_expired(self) -> int: + """Drop expired entries; return how many remain.""" + self._purge(self._clock()) + return len(self._seen) + + def size(self) -> int: + """Number of live (non-expired) entries.""" + return self.purge_expired() diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 9d527783..8216393c 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2930,6 +2930,16 @@ def _rate_limit(name: str, rate: float = 1.0, capacity: float = 1.0, _BULKHEADS: Dict[str, Any] = {} _IDEMPOTENCY_STORES: Dict[str, Any] = {} +_DEDUP_WINDOWS: Dict[str, Any] = {} + + +def _dedup_check(name: str, message_id: str, + ttl_s: Any = 3600) -> Dict[str, Any]: + """Adapter: check-and-mark a message id in a named dedup window.""" + from je_auto_control.utils.dedup_window import DedupWindow + window = _DEDUP_WINDOWS.setdefault(name, DedupWindow(float(ttl_s))) + return {"first_seen": window.check_and_mark(message_id), + "size": window.size()} def _idempotency_begin(name: str, key: str, @@ -4566,6 +4576,7 @@ def __init__(self): "AC_ewma": _ewma, "AC_idempotency_begin": _idempotency_begin, "AC_idempotency_complete": _idempotency_complete, + "AC_dedup_check": _dedup_check, "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 index b9885c01..523bec4b 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -3548,6 +3548,23 @@ def dataset_diff_tools() -> List[MCPTool]: ] +def dedup_window_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_dedup_check", + description=("Check-and-mark a 'message_id' in a named dedup window " + "'name' (TTL 'ttl_s'). Returns {first_seen, size} — " + "first_seen is false for a duplicate within the window."), + input_schema=schema( + {"name": {"type": "string"}, + "message_id": {"type": "string"}, "ttl_s": {"type": "number"}}, + ["name", "message_id"]), + handler=h.dedup_check, + annotations=NON_DESTRUCTIVE, + ), + ] + + def idempotency_tools() -> List[MCPTool]: return [ MCPTool( @@ -5553,6 +5570,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, 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 index 9b9a0cf2..df0d80d7 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1932,6 +1932,11 @@ def idempotency_complete(name, key, response): return _idempotency_complete(name, key, response) +def dedup_check(name, message_id, ttl_s=3600): + from je_auto_control.utils.executor.action_executor import _dedup_check + return _dedup_check(name, message_id, ttl_s) + + 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/test/unit_test/headless/test_dedup_window_batch.py b/test/unit_test/headless/test_dedup_window_batch.py new file mode 100644 index 00000000..88a1c6ba --- /dev/null +++ b/test/unit_test/headless/test_dedup_window_batch.py @@ -0,0 +1,64 @@ +"""Headless tests for the time-windowed dedup inbox. Pure stdlib, no Qt.""" +import je_auto_control as ac +from je_auto_control.utils.dedup_window import DedupWindow + + +def test_check_and_mark_first_then_duplicate(): + window = DedupWindow(60, clock=lambda: 1000.0) + assert window.check_and_mark("m1") is True # first + assert window.check_and_mark("m1") is False # duplicate + assert window.check_and_mark("m2") is True + + +def test_seen_and_mark(): + now = [0.0] + window = DedupWindow(10, clock=lambda: now[0]) + assert window.seen("x") is False + window.mark("x") + assert window.seen("x") is True + + +def test_ttl_eviction(): + now = [100.0] + window = DedupWindow(10, clock=lambda: now[0]) + window.mark("x") + assert window.seen("x") is True + now[0] += 20 # past TTL + assert window.seen("x") is False + assert window.check_and_mark("x") is True # re-allowed after expiry + + +def test_size_counts_live_only(): + now = [0.0] + window = DedupWindow(5, clock=lambda: now[0]) + window.mark("a") + window.mark("b") + assert window.size() == 2 + now[0] += 10 + assert window.size() == 0 + + +# --- wiring --------------------------------------------------------------- + +def test_executor_round_trip(): + name = "dedup-exec-test" + rec = ac.execute_action([[ + "AC_dedup_check", {"name": name, "message_id": "e1"}]]) + assert next(v for v in rec.values() + if isinstance(v, dict))["first_seen"] is True + rec2 = ac.execute_action([[ + "AC_dedup_check", {"name": name, "message_id": "e1"}]]) + assert next(v for v in rec2.values() + if isinstance(v, dict))["first_seen"] is False + + +def test_wiring(): + assert "AC_dedup_check" in ac.executor.known_commands() + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + assert "ac_dedup_check" in {t.name for t in build_default_tool_registry()} + from je_auto_control.gui.script_builder.command_schema import _build_specs + assert "AC_dedup_check" in {s.command for s in _build_specs()} + + +def test_facade_exports(): + assert hasattr(ac, "DedupWindow") and "DedupWindow" in ac.__all__