Add time-windowed message deduplication (exactly-once inbox)

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).

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)。

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)。

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**.

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

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** 分類下的命令提供。

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

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",

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=(

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"]

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()

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,

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,

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)

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__