Add per-stream sequence-gap detection

README.md

@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [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)
 - [What's new (2026-06-22) — Moving-Average Smoothing](#whats-new-2026-06-22--moving-average-smoothing)
@@ -157,6 +158,12 @@
 
 ---
 
+## 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).
+
+- **`SequenceTracker`** (`AC_sequence_observe`): nothing tracked per-stream monotonic sequence numbers. `observe(stream, seq)` classifies each as `ok` / `duplicate` / `gap` (with the `missing` numbers) / `reorder` (late arrivals fill gaps), and exposes `gaps` and `high_water`. Complements `dedup_window`. Pure-stdlib, deterministic.
+
 ## 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).

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--移动平均平滑)
@@ -160,6 +161,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/v105_features_doc.rst`](../docs/source/Zh/doc/new_features/v105_features_doc.rst)。
+
+- **`SequenceTracker`**(`AC_sequence_observe`):没有东西追踪每个流的单调序号。`observe(stream, seq)` 将每个分类为 `ok` / `duplicate` / `gap`(附 `missing` 序号)/ `reorder`(迟到填补间隙),并提供 `gaps` 与 `high_water`。与 `dedup_window` 互补。纯标准库、确定。
+
 ## 本次更新 (2026-06-22) — 时间窗口去重
 
 在 TTL 窗口内丢弃重复/重送的消息。完整参考:[`docs/source/Zh/doc/new_features/v104_features_doc.rst`](../docs/source/Zh/doc/new_features/v104_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--移動平均平滑)
@@ -160,6 +161,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/v105_features_doc.rst`](../docs/source/Zh/doc/new_features/v105_features_doc.rst)。
+
+- **`SequenceTracker`**(`AC_sequence_observe`):沒有東西追蹤每個串流的單調序號。`observe(stream, seq)` 將每個分類為 `ok` / `duplicate` / `gap`(附 `missing` 序號)/ `reorder`(遲到填補間隙),並提供 `gaps` 與 `high_water`。與 `dedup_window` 互補。純標準函式庫、具決定性。
+
 ## 本次更新 (2026-06-22) — 時間視窗去重
 
 在 TTL 視窗內丟棄重複/重送的訊息。完整參考:[`docs/source/Zh/doc/new_features/v104_features_doc.rst`](../docs/source/Zh/doc/new_features/v104_features_doc.rst)。

docs/source/Eng/doc/new_features/v105_features_doc.rst

@@ -0,0 +1,39 @@
+Per-Stream Sequence-Gap Detection
+=================================
+
+Nothing tracked per-stream monotonic sequence numbers to detect missing,
+out-of-order, or duplicate messages. ``dedup_window`` says "seen this id
+before"; this complements it by classifying each sequence number and tracking
+the outstanding gaps and high-water mark per stream.
+
+Pure standard library; imports no ``PySide6``. State is in-memory and fully
+injectable, so detection is deterministic in CI.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import SequenceTracker
+
+    tracker = SequenceTracker()
+    tracker.observe("orders", 1)      # {"status": "ok", ...}
+    tracker.observe("orders", 4)      # {"status": "gap", "missing": [2, 3]}
+    tracker.observe("orders", 3)      # {"status": "reorder", "missing": [2]}
+    tracker.gaps("orders")            # [2]
+    tracker.high_water("orders")      # 4
+
+``observe`` returns ``{status, seq, missing}`` where status is ``ok`` (next in
+order or the first seen), ``duplicate`` (already seen), ``gap`` (numbers were
+skipped — they are recorded as missing), or ``reorder`` (a late earlier number,
+which fills a gap when applicable). ``gaps`` lists the outstanding missing
+numbers and ``high_water`` is the highest seen. Streams are tracked
+independently by ``stream_id``.
+
+Executor command
+----------------
+
+``AC_sequence_observe`` observes a ``seq`` on a ``stream_id`` in a named tracker
+and returns the classification. It uses a named-instance registry and is exposed
+as the MCP tool ``ac_sequence_observe`` and as a Script Builder command under
+**Flow**.

docs/source/Eng/eng_index.rst

@@ -127,6 +127,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v102_features_doc
    doc/new_features/v103_features_doc
    doc/new_features/v104_features_doc
+   doc/new_features/v105_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

docs/source/Zh/doc/new_features/v105_features_doc.rst

@@ -0,0 +1,31 @@
+逐串流序號間隙偵測
+================
+
+沒有東西追蹤每個串流的單調序號以偵測遺漏、亂序或重複的訊息。``dedup_window`` 說「之前看過這個 id」;
+本功能補上互補的一面:分類每個序號,並追蹤每個串流的未決間隙與 high-water 標記。
+
+純標準函式庫;不匯入 ``PySide6``。狀態為記憶體內且完全可注入,因此偵測在 CI 中具決定性。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import SequenceTracker
+
+    tracker = SequenceTracker()
+    tracker.observe("orders", 1)      # {"status": "ok", ...}
+    tracker.observe("orders", 4)      # {"status": "gap", "missing": [2, 3]}
+    tracker.observe("orders", 3)      # {"status": "reorder", "missing": [2]}
+    tracker.gaps("orders")            # [2]
+    tracker.high_water("orders")      # 4
+
+``observe`` 回傳 ``{status, seq, missing}``,status 為 ``ok``(順序下一個或首次)、``duplicate``(已看過)、
+``gap``(序號被跳過 —— 記為遺漏)或 ``reorder``(較早的遲到序號,可填補間隙)。``gaps`` 列出未決遺漏序號,
+``high_water`` 為最高已見序號。各串流以 ``stream_id`` 獨立追蹤。
+
+執行器命令
+----------
+
+``AC_sequence_observe`` 在具名追蹤器中觀察某 ``stream_id`` 的 ``seq`` 並回傳分類。它使用具名實例登錄,
+並以 MCP 工具 ``ac_sequence_observe`` 以及 Script Builder 中 **Flow** 分類下的命令提供。

docs/source/Zh/zh_index.rst

@@ -127,6 +127,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v102_features_doc
    doc/new_features/v103_features_doc
    doc/new_features/v104_features_doc
+   doc/new_features/v105_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

je_auto_control/__init__.py

@@ -205,6 +205,8 @@
 )
 # Time-windowed message deduplication (exactly-once inbox)
 from je_auto_control.utils.dedup_window import DedupWindow
+# Per-stream sequence-gap / ordering detection
+from je_auto_control.utils.sequence_gap import SequenceTracker
 # CI workflow annotations (GitHub Actions)
 from je_auto_control.utils.ci_annotations import (
     emit_annotations, format_annotation,
@@ -932,7 +934,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "replay_timeline", "run_sequence",
     "CircuitBreaker", "CircuitOpenError", "RetryPolicy", "retry_call",
     "IdempotencyConflict", "IdempotencyStore", "request_fingerprint",
-    "DedupWindow",
+    "DedupWindow", "SequenceTracker",
     "emit_annotations", "format_annotation",
     "ClipboardHistory", "default_clipboard_history",
     "analyze_heal_log", "heal_stats", "scan_secrets",

je_auto_control/gui/script_builder/command_schema.py

@@ -2023,6 +2023,15 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Check-and-mark a message id; first_seen false on duplicate.",
     ))
+    specs.append(CommandSpec(
+        "AC_sequence_observe", "Flow", "Sequence: Observe",
+        fields=(
+            FieldSpec("name", FieldType.STRING, placeholder="ingest"),
+            FieldSpec("stream_id", FieldType.STRING, placeholder="orders"),
+            FieldSpec("seq", FieldType.INT, placeholder="14"),
+        ),
+        description="Classify a sequence number (ok/duplicate/gap/reorder).",
+    ))
     specs.append(CommandSpec(
         "AC_diff_rows", "Data", "Dataset Diff: Rows by Key",
         fields=(

je_auto_control/utils/executor/action_executor.py

@@ -2931,6 +2931,14 @@ 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] = {}
+_SEQUENCE_TRACKERS: Dict[str, Any] = {}
+
+
+def _sequence_observe(name: str, stream_id: str, seq: Any) -> Dict[str, Any]:
+    """Adapter: observe a sequence number in a named tracker."""
+    from je_auto_control.utils.sequence_gap import SequenceTracker
+    tracker = _SEQUENCE_TRACKERS.setdefault(name, SequenceTracker())
+    return tracker.observe(stream_id, int(seq))
 
 
 def _dedup_check(name: str, message_id: str,
@@ -4577,6 +4585,7 @@ def __init__(self):
             "AC_idempotency_begin": _idempotency_begin,
             "AC_idempotency_complete": _idempotency_complete,
             "AC_dedup_check": _dedup_check,
+            "AC_sequence_observe": _sequence_observe,
             "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 sequence_gap_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_sequence_observe",
+            description=("Observe sequence number 'seq' on 'stream_id' in named "
+                         "tracker 'name'. Returns {status: ok|duplicate|gap|"
+                         "reorder, seq, missing}."),
+            input_schema=schema(
+                {"name": {"type": "string"}, "stream_id": {"type": "string"},
+                 "seq": {"type": "integer"}},
+                ["name", "stream_id", "seq"]),
+            handler=h.sequence_observe,
+            annotations=NON_DESTRUCTIVE,
+        ),
+    ]
+
+
 def dedup_window_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -5570,7 +5587,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,
+    dedup_window_tools, sequence_gap_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

@@ -1937,6 +1937,11 @@ def dedup_check(name, message_id, ttl_s=3600):
     return _dedup_check(name, message_id, ttl_s)
 
 
+def sequence_observe(name, stream_id, seq):
+    from je_auto_control.utils.executor.action_executor import _sequence_observe
+    return _sequence_observe(name, stream_id, seq)
+
+
 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)

je_auto_control/utils/sequence_gap/__init__.py

@@ -0,0 +1,4 @@
+"""Per-stream sequence-gap detection for AutoControl."""
+from je_auto_control.utils.sequence_gap.sequence_gap import SequenceTracker
+
+__all__ = ["SequenceTracker"]

je_auto_control/utils/sequence_gap/sequence_gap.py

@@ -0,0 +1,61 @@
+"""Per-stream sequence-gap / ordering detection.
+
+Nothing tracked per-stream monotonic sequence numbers to detect missing,
+out-of-order, or duplicate messages. ``dedup_window`` says "seen this id
+before"; this complements it by classifying each sequence number as ``ok`` /
+``duplicate`` / ``gap`` (numbers skipped ahead) / ``reorder`` (a late arrival),
+and tracking the outstanding gaps and high-water mark per stream.
+
+Pure standard library; imports no ``PySide6``. State is in-memory and fully
+injectable, so detection is deterministic in CI.
+"""
+from typing import Any, Dict, List
+
+
+class SequenceTracker:
+    """Tracks per-stream sequence numbers and classifies each observation."""
+
+    def __init__(self) -> None:
+        self._streams: Dict[str, Dict[str, Any]] = {}
+
+    def _state(self, stream_id: str) -> Dict[str, Any]:
+        return self._streams.setdefault(
+            stream_id, {"high": None, "seen": set(), "gaps": set()})
+
+    def observe(self, stream_id: str, seq: int) -> Dict[str, Any]:
+        """Record ``seq`` for ``stream_id``; return its classification.
+
+        Returns ``{status, seq, missing}`` where status is ``ok`` (next in
+        order), ``duplicate`` (seen before), ``gap`` (skipped ahead), or
+        ``reorder`` (a late earlier number). ``missing`` is the outstanding
+        gap list.
+        """
+        state = self._state(stream_id)
+        seq = int(seq)
+        if seq in state["seen"]:
+            return {"status": "duplicate", "seq": seq,
+                    "missing": sorted(state["gaps"])}
+        state["seen"].add(seq)
+        status = self._advance(state, seq)
+        return {"status": status, "seq": seq, "missing": sorted(state["gaps"])}
+
+    @staticmethod
+    def _advance(state: Dict[str, Any], seq: int) -> str:
+        high = state["high"]
+        if high is None or seq == high + 1:
+            state["high"] = seq
+            return "ok"
+        if seq > high + 1:
+            state["gaps"].update(range(high + 1, seq))
+            state["high"] = seq
+            return "gap"
+        state["gaps"].discard(seq)          # a late arrival filling a hole
+        return "reorder"
+
+    def gaps(self, stream_id: str) -> List[int]:
+        """The outstanding missing sequence numbers for ``stream_id``."""
+        return sorted(self._state(stream_id)["gaps"])
+
+    def high_water(self, stream_id: str) -> Any:
+        """The highest sequence number observed for ``stream_id`` (or ``None``)."""
+        return self._state(stream_id)["high"]

test/unit_test/headless/test_sequence_gap_batch.py

@@ -0,0 +1,60 @@
+"""Headless tests for per-stream sequence-gap detection. No Qt."""
+import je_auto_control as ac
+from je_auto_control.utils.sequence_gap import SequenceTracker
+
+
+def test_in_order_is_ok():
+    t = SequenceTracker()
+    assert t.observe("s", 1)["status"] == "ok"
+    assert t.observe("s", 2)["status"] == "ok"
+    assert t.observe("s", 3)["status"] == "ok"
+    assert t.gaps("s") == [] and t.high_water("s") == 3
+
+
+def test_gap_detected_and_filled_by_reorder():
+    t = SequenceTracker()
+    t.observe("s", 1)
+    result = t.observe("s", 4)               # skipped 2, 3
+    assert result["status"] == "gap" and result["missing"] == [2, 3]
+    assert t.observe("s", 3)["status"] == "reorder"
+    assert t.gaps("s") == [2]                # 3 filled, 2 still missing
+    t.observe("s", 2)
+    assert t.gaps("s") == []
+
+
+def test_duplicate():
+    t = SequenceTracker()
+    t.observe("s", 1)
+    t.observe("s", 2)
+    assert t.observe("s", 2)["status"] == "duplicate"
+
+
+def test_streams_are_independent():
+    t = SequenceTracker()
+    t.observe("a", 5)
+    assert t.observe("b", 1)["status"] == "ok"      # first on stream b
+    assert t.high_water("a") == 5 and t.high_water("b") == 1
+
+
+# --- wiring ---------------------------------------------------------------
+
+def test_executor_round_trip():
+    name = "seq-exec-test"
+    ac.execute_action([[
+        "AC_sequence_observe", {"name": name, "stream_id": "s", "seq": 1}]])
+    rec = ac.execute_action([[
+        "AC_sequence_observe", {"name": name, "stream_id": "s", "seq": 5}]])
+    out = next(v for v in rec.values() if isinstance(v, dict))
+    assert out["status"] == "gap" and out["missing"] == [2, 3, 4]
+
+
+def test_wiring():
+    assert "AC_sequence_observe" in ac.executor.known_commands()
+    from je_auto_control.utils.mcp_server.tools import build_default_tool_registry
+    assert "ac_sequence_observe" in {t.name for t in build_default_tool_registry()}
+    from je_auto_control.gui.script_builder.command_schema import _build_specs
+    assert "AC_sequence_observe" in {s.command for s in _build_specs()}
+
+
+def test_facade_exports():
+    assert hasattr(ac, "SequenceTracker") and "SequenceTracker" in ac.__all__