From 5fc2c967e4284434024e6e34b4d31dac2dcec8ff Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Sat, 17 Oct 2026 04:43:37 +0800
Subject: [PATCH] refactor(types): describe sync and async debounce contracts

---
 app/runtime/debounce.py         | 120 +++++++++++++--------
 tests/test_debounce_contract.py | 184 ++++++++++++++++++++++++++++++++
 2 files changed, 261 insertions(+), 43 deletions(-)
 create mode 100644 tests/test_debounce_contract.py

diff --git a/app/runtime/debounce.py b/app/runtime/debounce.py
index b7a0aea63..e368e3ed5 100644
--- a/app/runtime/debounce.py
+++ b/app/runtime/debounce.py
@@ -3,19 +3,45 @@ import functools
 import inspect
 from abc import ABC, abstractmethod
 from threading import Lock, Timer
-from typing import Any, Callable, Optional
+from typing import Awaitable, Callable, Concatenate, Generic, Optional, ParamSpec, Protocol, TypeVar, cast, overload
 
 from app.runtime.log import logger
 
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+_T = TypeVar("_T")
 
-class BaseDebouncer(ABC):
+
+class _Debounced(Protocol[_P]):
+    """保留调用参数；实际函数种类决定调用与取消是否返回可等待对象。"""
+
+    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> Awaitable[None] | None:
+        """提交本次调用；同步函数直接返回，协程函数需要等待提交。"""
+        ...
+
+    # 包装函数挂载的控制方法，异步引擎会等待任务进入终态。
+    cancel: Callable[[], Awaitable[None] | None]
+
+    @overload
+    def __get__(self, instance: None, owner: type[object] | None = None) -> "_Debounced[_P]":
+        """从类访问时保留尚未绑定的参数。"""
+        ...
+
+    @overload
+    def __get__(self: "_Debounced[Concatenate[_T, _P]]", instance: _T,
+                owner: type[_T] | None = None) -> "_Debounced[_P]":
+        """从实例访问时由函数描述符绑定首个参数。"""
+        ...
+
+
+class BaseDebouncer(ABC, Generic[_P, _R]):
     """
     防抖器的抽象基类。定义了防抖器的基本接口和日志功能。
     所有防抖器实现类必须继承此类并实现其抽象方法。
     """
 
-    def __init__(self, func: Callable, interval: float, *,
-                 leading: bool = False, enable_logging: bool = False, source: str = ""):
+    def __init__(self, func: Callable[_P, _R], interval: float, *,
+                 leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
         """
         初始化防抖器实例。
         :param func: 要防抖的函数或协程
@@ -31,14 +57,14 @@ class BaseDebouncer(ABC):
         self.source = source
 
     @abstractmethod
-    def __call__(self, *args, **kwargs) -> None:
+    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> Awaitable[None] | None:
         """
         定义防抖调用的契约，子类必须实现。
         """
         pass
 
     @abstractmethod
-    def cancel(self) -> None:
+    def cancel(self) -> Awaitable[None] | None:
         """
         定义取消挂起调用的契约，子类必须实现。
         """
@@ -50,7 +76,7 @@ class BaseDebouncer(ABC):
         """
         return f"[{self.source}] {message}" if self.source else message
 
-    def log(self, level: str, message: str):
+    def log(self, level: str, message: str) -> None:
         """
         根据日志级别记录日志。
         """
@@ -58,53 +84,54 @@ class BaseDebouncer(ABC):
             log_method = getattr(logger, level, logger.debug)
             log_method(self.format_log(message))
 
-    def log_debug(self, message: str):
+    def log_debug(self, message: str) -> None:
         """
         记录调试日志。
         """
         self.log("debug", message)
 
-    def log_info(self, message: str):
+    def log_info(self, message: str) -> None:
         """
         记录信息日志。
         """
         self.log("info", message)
 
-    def log_warning(self, message: str):
+    def log_warning(self, message: str) -> None:
         """
         记录警告日志。
         """
         self.log("warning", message)
 
-    def error(self, message: str):
+    def error(self, message: str) -> None:
         """
         记录错误日志。
         """
         self.log("error", message)
 
-    def critical(self, message: str):
+    def critical(self, message: str) -> None:
         """
         记录严重错误日志。
         """
         self.log("critical", message)
 
 
-class Debouncer(BaseDebouncer):
+class Debouncer(BaseDebouncer[_P, object]):
     """
     同步防抖实现类
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, func: Callable[_P, object], interval: float, *,
+                 leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
         """
         初始化防抖器实例。
         """
-        super().__init__(*args, **kwargs)
+        super().__init__(func, interval, leading=leading, enable_logging=enable_logging, source=source)
         self.timer: Optional[Timer] = None
         self.lock = Lock()
         # 用于前沿模式，标记是否处于“冷却”或“不应期”
         self.is_cooling_down = False
 
-    def __call__(self, *args, **kwargs) -> None:
+    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         调用防抖函数。
         :param args:
@@ -117,7 +144,7 @@ class Debouncer(BaseDebouncer):
             else:
                 self._call_trailing(*args, **kwargs)
 
-    def _call_leading(self, *args, **kwargs):
+    def _call_leading(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         前沿模式的逻辑。
         """
@@ -138,7 +165,7 @@ class Debouncer(BaseDebouncer):
         self.timer.start()
         self.log_debug(f"前沿模式: 进入 {self.interval} 秒的冷却期。")
 
-    def _end_cool_down(self):
+    def _end_cool_down(self) -> None:
         """
         计时器到期后，解除冷却状态
         """
@@ -146,7 +173,7 @@ class Debouncer(BaseDebouncer):
             self.is_cooling_down = False
             self.log_debug("前沿模式: 冷却时间结束，可以再次立即执行。")
 
-    def _call_trailing(self, *args, **kwargs):
+    def _call_trailing(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         后沿模式的逻辑。
         """
@@ -155,7 +182,8 @@ class Debouncer(BaseDebouncer):
             self.timer.cancel()
             self.log_debug("后沿模式: 检测到新的调用，已重置计时器。")
 
-        def execute():
+        def execute() -> None:
+            """计时结束后执行最后一次调用保存的参数。"""
             self.log_info("后沿模式: 计时结束，开始执行函数。")
             self.func(*args, **kwargs)
 
@@ -175,18 +203,19 @@ class Debouncer(BaseDebouncer):
             self.is_cooling_down = False
 
 
-class AsyncDebouncer(BaseDebouncer):
+class AsyncDebouncer(BaseDebouncer[_P, Awaitable[object]]):
     """
     异步防抖实现类。
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, func: Callable[_P, Awaitable[object]], interval: float, *,
+                 leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
         """
         初始化异步防抖器实例。
         """
-        super().__init__(*args, **kwargs)
-        self.task: Optional[asyncio.Task] = None
-        self._retired_tasks: set[asyncio.Task[Any]] = set()
+        super().__init__(func, interval, leading=leading, enable_logging=enable_logging, source=source)
+        self.task: Optional[asyncio.Task[None]] = None
+        self._retired_tasks: set[asyncio.Task[None]] = set()
         self.lock = asyncio.Lock()
         self.is_cooling_down = False
 
@@ -199,7 +228,7 @@ class AsyncDebouncer(BaseDebouncer):
         self._retired_tasks.add(task)
         task.add_done_callback(self._retired_tasks.discard)
 
-    async def __call__(self, *args, **kwargs) -> None:
+    async def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         异步调用防抖函数。
         """
@@ -209,7 +238,7 @@ class AsyncDebouncer(BaseDebouncer):
             else:
                 await self._call_trailing(*args, **kwargs)
 
-    async def _call_leading(self, *args, **kwargs):
+    async def _call_leading(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         前沿模式的逻辑。
         """
@@ -223,7 +252,7 @@ class AsyncDebouncer(BaseDebouncer):
         self.task = asyncio.create_task(self._end_cool_down())
         self.log_debug(f"前沿模式 (async): 进入 {self.interval} 秒的冷却期。")
 
-    async def _end_cool_down(self):
+    async def _end_cool_down(self) -> None:
         """
         计时器到期后，解除冷却状态
         """
@@ -232,7 +261,7 @@ class AsyncDebouncer(BaseDebouncer):
             self.is_cooling_down = False
             self.log_debug("前沿模式 (async): 冷却时间结束。")
 
-    async def _call_trailing(self, *args, **kwargs):
+    async def _call_trailing(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         后沿模式的逻辑。
         """
@@ -243,7 +272,7 @@ class AsyncDebouncer(BaseDebouncer):
         self.task = asyncio.create_task(self._delayed_execute(*args, **kwargs))
         self.log_debug(f"后沿模式 (async): 任务已创建，将在 {self.interval} 秒后执行。")
 
-    async def _delayed_execute(self, *args, **kwargs):
+    async def _delayed_execute(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
         """
         延迟执行实际的协程函数。
         """
@@ -279,26 +308,29 @@ class AsyncDebouncer(BaseDebouncer):
 
 
 def debounce(interval: float, *, leading: bool = False,
-             enable_logging: bool = False, source: str = "") -> Callable:
+             enable_logging: bool = False, source: str = "") -> Callable[[Callable[_P, object]], _Debounced[_P]]:
     """
     支持同步和异步的防抖装饰器工厂。
     """
 
-    def decorator(func: Callable) -> Callable:
+    def decorator(func: Callable[_P, object]) -> _Debounced[_P]:
+        """按实际协程函数标记选择引擎，保留原函数的参数与绑定行为。"""
         # 检查函数类型，并选择合适的引擎
         if inspect.iscoroutinefunction(func):
             # 异步函数，使用 AsyncDebouncer
-            instance = AsyncDebouncer(func, interval,
-                                      leading=leading,
-                                      enable_logging=enable_logging,
-                                      source=source)
+            async_instance = AsyncDebouncer(func, interval,
+                                            leading=leading,
+                                            enable_logging=enable_logging,
+                                            source=source)
 
             @functools.wraps(func)
-            async def async_wrapper(*args, **kwargs) -> Any:
-                await instance(*args, **kwargs)
+            async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
+                """等待异步引擎接收本次调用。"""
+                await async_instance(*args, **kwargs)
 
-            async_wrapper.cancel = instance.cancel
-            return async_wrapper
+            async_result = cast(_Debounced[_P], async_wrapper)
+            async_result.cancel = async_instance.cancel
+            return async_result
 
         else:
             # 同步函数，使用 Debouncer
@@ -308,10 +340,12 @@ def debounce(interval: float, *, leading: bool = False,
                                  source=source)
 
             @functools.wraps(func)
-            def wrapper(*args, **kwargs) -> Any:
+            def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
+                """通过同步引擎接收本次调用。"""
                 instance(*args, **kwargs)
 
-            wrapper.cancel = instance.cancel
-            return wrapper
+            result = cast(_Debounced[_P], wrapper)
+            result.cancel = instance.cancel
+            return result
 
     return decorator
diff --git a/tests/test_debounce_contract.py b/tests/test_debounce_contract.py
new file mode 100644
index 000000000..4e7c5e2c4
--- /dev/null
+++ b/tests/test_debounce_contract.py
@@ -0,0 +1,184 @@
+"""防抖器参数转发、函数绑定与同步异步选择的兼容契约。"""
+
+import asyncio
+import inspect
+
+import pytest
+
+import app.runtime.debounce as debounce_module
+from app.runtime.debounce import AsyncDebouncer, Debouncer, debounce
+
+
+class _ManualTimer:
+    """由测试显式推进的定时器，避免线程与墙钟时间影响断言。"""
+
+    def __init__(self, interval, callback):
+        """保存真实引擎提供的间隔与回调。"""
+        self.interval = interval
+        self.callback = callback
+        self.active = False
+
+    def start(self):
+        """标记为待执行。"""
+        self.active = True
+
+    def cancel(self):
+        """撤销尚未执行的回调。"""
+        self.active = False
+
+    def is_alive(self):
+        """提供与线程定时器一致的存活查询。"""
+        return self.active
+
+    def fire(self):
+        """仅执行仍有效的回调。"""
+        if self.active:
+            self.active = False
+            self.callback()
+
+
+@pytest.fixture
+def manual_timers(monkeypatch):
+    """捕获引擎创建的定时器并禁止启动真实线程。"""
+    timers = []
+
+    def create_timer(interval, callback):
+        """返回可由测试控制的独立定时器。"""
+        timer = _ManualTimer(interval, callback)
+        timers.append(timer)
+        return timer
+
+    monkeypatch.setattr(debounce_module, "Timer", create_timer)
+    return timers
+
+
+@pytest.mark.parametrize("keyword_constructor", [False, True])
+@pytest.mark.parametrize("leading", [False, True])
+def test_sync_debouncer_preserves_arguments_and_cooling(
+    manual_timers, keyword_constructor, leading
+):
+    """两种构造方式均保留参数，并分别执行首个调用或最后一个调用。"""
+    calls = []
+
+    def record(value, *, label):
+        """保留位置参数、关键字参数和对象身份。"""
+        calls.append((value, label))
+        return "discarded"
+
+    options = {"leading": leading, "enable_logging": False, "source": "test"}
+    engine = (
+        Debouncer(func=record, interval=7, **options)
+        if keyword_constructor
+        else Debouncer(record, 7, **options)
+    )
+    first, last = object(), object()
+    assert engine(first, label="first") is None
+    assert engine(last, label="last") is None
+    assert manual_timers[0].is_alive() is False
+    assert manual_timers[-1].interval == 7
+    assert calls == ([(first, "first")] if leading else [])
+    for timer in manual_timers:
+        timer.fire()
+    assert calls == ([(first, "first")] if leading else [(last, "last")])
+    assert engine.is_cooling_down is False
+    engine(first, label="cancelled")
+    before_cancel = list(calls)
+    assert engine.cancel() is None
+    manual_timers[-1].fire()
+    assert calls == before_cancel
+    assert engine.timer is None
+
+
+def test_decorator_preserves_method_binding_and_metadata(manual_timers):
+    """普通函数描述符仍绑定实例，cancel 仍操作原先共享的防抖引擎。"""
+    calls = []
+
+    class Receiver:
+        """通过实例方法暴露防抖包装器。"""
+
+        @debounce(3)
+        def accept(self, value, *, label):
+            """记录接收实例与完整调用参数。"""
+            calls.append((self, value, label))
+
+    receiver = Receiver()
+    assert receiver.accept("value", label="label") is None
+    assert Receiver.accept.__name__ == "accept"
+    assert inspect.iscoroutinefunction(receiver.accept) is False
+    manual_timers[-1].fire()
+    assert calls == [(receiver, "value", "label")]
+    receiver.accept("cancelled", label="label")
+    assert receiver.accept.cancel() is None
+    manual_timers[-1].fire()
+    assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("leading", [False, True])
+async def test_async_debouncer_preserves_arguments_and_result_discard(leading):
+    """协程函数通过异步引擎执行，调用和取消均保持返回 None 的约定。"""
+    calls = []
+
+    async def record(value, *, label):
+        """记录实际被执行的调用。"""
+        calls.append((value, label))
+        return "discarded"
+
+    engine = AsyncDebouncer(func=record, interval=0, leading=leading, source="test")
+    assert await engine("first", label=1) is None
+    assert await engine("last", label=2) is None
+    await asyncio.wait_for(engine.task, timeout=1)
+    assert calls == ([("first", 1)] if leading else [("last", 2)])
+    assert await engine.cancel() is None
+    assert engine.task is None
+    assert engine._retired_tasks == set()
+
+
+@pytest.mark.asyncio
+async def test_async_decorator_preserves_binding_and_cancel():
+    """异步方法包装保留 await、实例绑定、元数据和异步取消接口。"""
+    calls = []
+
+    class Receiver:
+        """提供异步防抖方法。"""
+
+        @debounce(60, leading=True)
+        async def accept(self, value, *, label):
+            """记录绑定后的调用并返回被包装器丢弃的结果。"""
+            calls.append((self, value, label))
+            return "discarded"
+
+    receiver = Receiver()
+    try:
+        assert inspect.iscoroutinefunction(receiver.accept)
+        assert receiver.accept.__name__ == "accept"
+        assert await receiver.accept("value", label="label") is None
+        assert calls == [(receiver, "value", "label")]
+    finally:
+        assert await receiver.accept.cancel() is None
+
+
+def test_sync_function_returning_awaitable_stays_sync(manual_timers):
+    """返回可等待对象的普通函数仍走同步引擎，不能仅凭返回类型改选异步引擎。"""
+    events = []
+
+    class Result:
+        """记录是否有人等待了普通函数的返回值。"""
+
+        def __await__(self):
+            """提供不会实际等待的 await 协议。"""
+            events.append("awaited")
+            yield from ()
+            return "result"
+
+    @debounce(3, leading=True)
+    def callback(value):
+        """返回可等待对象但保持普通函数定义。"""
+        events.append(value)
+        return Result()
+
+    assert inspect.iscoroutinefunction(callback) is False
+    assert callback("called") is None
+    assert events == ["called"]
+    assert callback.cancel() is None
+    assert manual_timers[-1].is_alive() is False