diff --git a/README.md b/README.md index 5faa668e..a2255f71 100644 --- a/README.md +++ b/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). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index f91690ee..9056d406 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--移动平均平滑) @@ -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)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 49ab3ce4..bbe24dcc 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--移動平均平滑) @@ -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)。 diff --git a/docs/source/Eng/doc/new_features/v105_features_doc.rst b/docs/source/Eng/doc/new_features/v105_features_doc.rst new file mode 100644 index 00000000..987cf4b8 --- /dev/null +++ b/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**. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 6ba95b4c..74fa8dd0 100644 --- a/docs/source/Eng/eng_index.rst +++ b/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 diff --git a/docs/source/Zh/doc/new_features/v105_features_doc.rst b/docs/source/Zh/doc/new_features/v105_features_doc.rst new file mode 100644 index 00000000..512055d6 --- /dev/null +++ b/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** 分類下的命令提供。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index e0d936f3..0eaeb305 100644 --- a/docs/source/Zh/zh_index.rst +++ b/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 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index cb01f5bf..bc13d61f 100644 --- a/je_auto_control/__init__.py +++ b/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", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 204e41ca..35ad13f9 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/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=( diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 8216393c..85873166 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/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, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 523bec4b..d1a39757 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 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, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index df0d80d7..eb898c9f 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/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) diff --git a/je_auto_control/utils/sequence_gap/__init__.py b/je_auto_control/utils/sequence_gap/__init__.py new file mode 100644 index 00000000..788230eb --- /dev/null +++ b/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"] diff --git a/je_auto_control/utils/sequence_gap/sequence_gap.py b/je_auto_control/utils/sequence_gap/sequence_gap.py new file mode 100644 index 00000000..374b6feb --- /dev/null +++ b/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"] diff --git a/test/unit_test/headless/test_sequence_gap_batch.py b/test/unit_test/headless/test_sequence_gap_batch.py new file mode 100644 index 00000000..298efb77 --- /dev/null +++ b/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__