refactor(types): type singleton construction and cached instances

app/foundation/singleton.py

@@ -1,7 +1,11 @@
 import abc
 import threading
 import weakref
-from typing import Any
+from typing import Any, Callable, Optional, TypeVar, cast
+
+_ExistingInstance = TypeVar("_ExistingInstance")
+
+_InstanceKey = tuple[type[object], tuple[object, ...], frozenset[tuple[str, object]]]
 
 
 class Singleton(abc.ABCMeta, type):
@@ -9,7 +13,7 @@ class Singleton(abc.ABCMeta, type):
     类单例模式（按参数）
     """
 
-    _instances: dict = {}
+    _instances: dict[_InstanceKey, object] = {}
     _lock = threading.RLock()
 
     def get_existing_instance(cls, *args: Any, **kwargs: Any) -> Any:
@@ -30,7 +34,7 @@ class Singleton(abc.ABCMeta, type):
                 cls._instances.pop(key, None)
             return bool(keys)
 
-    def __call__(cls, *args, **kwargs):
+    def __call__(cls, *args: object, **kwargs: object) -> object:
         """按类和构造参数创建或复用实例。"""
         key = (cls, args, frozenset(kwargs.items()))
         with cls._lock:
@@ -38,11 +42,14 @@ class Singleton(abc.ABCMeta, type):
                 if getattr(cls, "_retain_failed_singleton", False):
                     # 启动线程的 lifecycle owner 必须先发布身份再执行 __init__；
                     # 构造中途抛错时保留实例，启动失败清理才能找到已创建的 owner。
-                    instance = cls.__new__(cls, *args, **kwargs)
+                    instance_type: type[object] = cls
+                    construct: Callable[..., object] = instance_type.__new__
+                    instance = construct(cls, *args, **kwargs)
                     if not isinstance(instance, cls):
                         return instance
                     cls._instances[key] = instance
-                    cls.__init__(instance, *args, **kwargs)
+                    initialize: Callable[..., None] = instance_type.__init__
+                    initialize(instance, *args, **kwargs)
                 else:
                     cls._instances[key] = super().__call__(*args, **kwargs)
             return cls._instances[key]
@@ -60,13 +67,15 @@ class SingletonClass(abc.ABCMeta, type):
     类单例模式（按类）
     """
 
-    _instances: dict = {}
+    _instances: dict[type[object], object] = {}
     _lock = threading.RLock()
 
-    def get_existing_instance(cls):
-        """返回已创建实例，不触发初始化"""
-        with cls._lock:
-            return cls._instances.get(cls)
+    def get_existing_instance(cls: type[_ExistingInstance]) -> Optional[_ExistingInstance]:
+        """返回与类键对应的已创建实例，不触发初始化。"""
+        # cls 是本元类的实例；缓存按该类键保存其构造结果。
+        singleton_class = cast(SingletonClass, cls)
+        with singleton_class._lock:
+            return cast(Optional[_ExistingInstance], singleton_class._instances.get(cls))
 
     def release_existing_instance(cls, instance: object) -> bool:
         """仅在身份匹配时释放已收敛实例，供显式生命周期重新创建 owner。"""
@@ -76,18 +85,21 @@ class SingletonClass(abc.ABCMeta, type):
             cls._instances.pop(cls, None)
             return True
 
-    def __call__(cls, *args, **kwargs):
+    def __call__(cls, *args: object, **kwargs: object) -> object:
         """按类创建或复用唯一实例。"""
         with cls._lock:
             if cls not in cls._instances:
                 if getattr(cls, "_retain_failed_singleton", False):
                     # 与参数化单例保持相同的 owner 发布顺序；锁会阻止其他线程
                     # 在 __init__ 返回或抛错前读取半构造实例。
-                    instance = cls.__new__(cls, *args, **kwargs)
+                    instance_type: type[object] = cls
+                    construct: Callable[..., object] = instance_type.__new__
+                    instance = construct(cls, *args, **kwargs)
                     if not isinstance(instance, cls):
                         return instance
                     cls._instances[cls] = instance
-                    cls.__init__(instance, *args, **kwargs)
+                    initialize: Callable[..., None] = instance_type.__init__
+                    initialize(instance, *args, **kwargs)
                 else:
                     cls._instances[cls] = super().__call__(*args, **kwargs)
             return cls._instances[cls]
@@ -104,7 +116,7 @@ class WeakSingleton(abc.ABCMeta, type):
     """
     弱引用单例模式 - 当没有强引用时自动清理
     """
-    _instances: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary()
+    _instances: weakref.WeakKeyDictionary[type[object], object] = weakref.WeakKeyDictionary()
     _lock = threading.RLock()
 
     def get_existing_instance(cls) -> Any:
@@ -120,7 +132,7 @@ class WeakSingleton(abc.ABCMeta, type):
             cls._instances.pop(cls, None)
             return True
 
-    def __call__(cls, *args, **kwargs):
+    def __call__(cls, *args: object, **kwargs: object) -> object:
         """按类创建或复用仍有强引用的实例。"""
         with cls._lock:
             if cls not in cls._instances:

docs/refactor/optimization-checklist.md

@@ -106,9 +106,9 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
 | Module Contract | 215 specs / 211 methods / 263 calls | 动态方法名为 0；分类事实富化通过显式合同进入插件调度，旧 YAML 分类方法已移除；download 新增 custom_words 可选参数，旧 provider 签名通过显式合同兼容 |
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
-| Python 源码量 | 371,953 行 | 本次重新扫描物理行数，排除 `app/plugins/**`；74 个文件超过 1,000 行，12 个超过 2,000 行；源码尺寸仅作观察 |
+| Python 源码量 | 371,965 行 | 本次重新扫描物理行数，排除 `app/plugins/**`；74 个文件超过 1,000 行，12 个超过 2,000 行；源码尺寸仅作观察 |
 | 长方法 | 309 个超过 80 行 | 本次 AST 扫描排除 `app/plugins/**`；61 个超过 150 行，17 个超过 250 行；新增函数仍受 C901/嵌套深度门禁约束 |
-| 全量 mypy 历史债务 | 9,062 / 493 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项，逐文件逐错误码无增长 |
+| 全量 mypy 历史债务 | 9,048 / 492 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项，逐文件逐错误码无增长 |
 | Ruff 历史诊断 | 100 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 

tests/fixtures/architecture/mypy-baseline.json

@@ -651,7 +651,7 @@
     "dict-item": 1,
     "misc": 1,
     "no-any-return": 4,
-    "no-untyped-call": 3,
+    "no-untyped-call": 2,
     "no-untyped-def": 12,
     "type-arg": 18,
     "union-attr": 2,
@@ -1285,13 +1285,6 @@
     "assignment": 2,
     "no-untyped-def": 5
   },
-  "app/foundation/singleton.py": {
-    "arg-type": 2,
-    "misc": 2,
-    "no-untyped-def": 4,
-    "type-arg": 3,
-    "var-annotated": 2
-  },
   "app/foundation/temporal.py": {
     "import-untyped": 3,
     "no-any-return": 1,

tests/test_singleton.py

@@ -7,7 +7,7 @@ def test_singleton_class_can_read_existing_instance_without_creating(monkeypatch
     """按类单例可以只读取已存在实例"""
 
     class Example(metaclass=SingletonClass):
-        pass
+        """只在显式构造后产生可查询实例的按类单例。"""
 
     monkeypatch.setattr(SingletonClass, "_instances", {})
 
@@ -20,7 +20,10 @@ def test_parameterized_singleton_can_read_matching_instance_without_creating(mon
     """参数化单例按相同参数读取已存在实例"""
 
     class Example(metaclass=Singleton):
+        """按名称区分实例的参数化单例。"""
+
         def __init__(self, name):
+            """保存用于区分实例的构造参数。"""
             self.name = name
 
     monkeypatch.setattr(Singleton, "_instances", {})
@@ -121,3 +124,62 @@ def test_class_singleton_can_retain_failed_lifecycle_owner(monkeypatch):
     retained = Example.get_existing_instance()
     assert retained is not None
     assert retained.owner_started is True
+
+
+@pytest.mark.parametrize("metaclass", [Singleton, SingletonClass])
+@pytest.mark.parametrize("retain_failed", [False, True])
+def test_singleton_forwards_constructor_arguments_once(monkeypatch, metaclass, retain_failed):
+    """普通与提前发布路径均应按原顺序向构造方法传递位置和关键字参数。"""
+    calls = []
+    monkeypatch.setattr(metaclass, "_instances", {})
+
+    class Example(metaclass=metaclass):
+        """记录自定义构造方法及初始化方法收到的参数。"""
+
+        _retain_failed_singleton = retain_failed
+
+        def __new__(cls, name, *, enabled):
+            """记录参数并创建当前类的实例。"""
+            calls.append(("new", name, enabled))
+            return object.__new__(cls)
+
+        def __init__(self, name, *, enabled):
+            """记录参数并检查需要保留的 owner 已提前发布。"""
+            calls.append(("init", name, enabled))
+            if retain_failed:
+                if metaclass is Singleton:
+                    assert Example.get_existing_instance(name, enabled=enabled) is self
+                else:
+                    assert Example.get_existing_instance() is self
+
+    instance = Example("owner", enabled=True)
+    assert Example("owner", enabled=True) is instance
+    assert calls == [("new", "owner", True), ("init", "owner", True)]
+
+
+@pytest.mark.parametrize("metaclass", [Singleton, SingletonClass])
+@pytest.mark.parametrize("retain_failed", [False, True])
+def test_singleton_preserves_foreign_new_result(monkeypatch, metaclass, retain_failed):
+    """自定义 __new__ 返回其它类型时跳过初始化，并保留原有缓存差异。"""
+    result = object()
+    calls = []
+    monkeypatch.setattr(metaclass, "_instances", {})
+
+    class Example(metaclass=metaclass):
+        """模拟返回异类对象的动态构造器。"""
+
+        _retain_failed_singleton = retain_failed
+
+        def __new__(cls):
+            """返回外部对象，验证元类没有强制转换或重建实例。"""
+            calls.append("new")
+            return result
+
+        def __init__(self):
+            """异类构造结果不应进入本类初始化方法。"""
+            raise AssertionError("foreign instance must not be initialized")
+
+    assert Example() is result
+    assert Example.get_existing_instance() is (None if retain_failed else result)
+    assert Example() is result
+    assert calls == ["new"] * (2 if retain_failed else 1)