fix(doeff-vm): EffectBase.__reduce_ex__ の copyreg 解決を PyOnceLock で 1 回きりにする

.semgrep.yaml

@@ -1401,6 +1401,29 @@ rules:
       include:
         - /packages/doeff-vm/src/pyvm.rs
 
+  - id: doeff-vm-no-per-call-copyreg-resolution
+    languages: [rust]
+    severity: ERROR
+    message: |
+      BANNED: resolving copyreg.__newobj__ per call (py.import("copyreg") /
+      getattr("__newobj__")). pickle hooks run on every effect instance, and
+      on free-threaded CPython 3.14 both import and module attribute lookup
+      take per-object locks — so a per-call resolution turns pickling into a
+      contention point. Measured 2026-08-07 in a long-lived multi-threaded
+      runtime: 948 threads parked on the import lock
+      (_PyMutex_LockTimed → _PyParkingLot_Park), reached from
+      PyImport_ImportModuleLevelObject → import_ensure_initialized and
+      _Py_module_getattro_impl, doeff_vm as the caller; 19.5 GiB paged out.
+      Resolve once into a static PyOnceLock — see COPYREG_NEWOBJ in
+      packages/doeff-vm/src/python_generator_stream.rs.
+    pattern-either:
+      - pattern-regex: py\.import\("copyreg"\)
+      - pattern-regex: getattr\("__newobj__"\)
+    paths:
+      include:
+        - "**/packages/doeff-vm/src/*.rs"
+        - "**/packages/doeff-vm-core/src/*.rs"
+
   # tombstone(2026-07-14): 退役済みパターン no-is-instance-from-in-handlers の再発防止として意図的に維持。現行の非fixture対象ゼロは想定内。
   - id: no-is-instance-from-in-handlers
     metadata:

docs/adr/enforcement-ledger.json

@@ -1,7 +1,7 @@
 {
   "_comment": "ADR-DOE-ENFORCE-001 R5 anti-drop ratchet の台帳。enforcement 資産の数が黙って減る(または黙って増える)ことを tests/test_enforcement_ledger.py が禁止する。数を変える変更は、この台帳の明示的な更新を同じ変更セットに含めること。",
   "defadr_files": 22,
-  "semgrep_rules": 247,
+  "semgrep_rules": 248,
   "adr_deftest_enforcements": 30,
   "adr_defsemgrep_enforcements": 45,
   "adr_laws": 72

packages/doeff-vm/src/python_generator_stream.rs

@@ -8,6 +8,7 @@
 use pyo3::exceptions::PyStopIteration;
 use pyo3::prelude::*;
 use pyo3::pyclass::{PyTraverseError, PyVisit};
+use pyo3::sync::PyOnceLock;
 use pyo3::types::PyString;
 
 use doeff_vm_core::do_ctrl::DoCtrl;
@@ -34,6 +35,20 @@ use doeff_vm_core::value::Value;
 #[derive(Debug)]
 pub struct PyEffectBase;
 
+/// `copyreg.__newobj__`, resolved once per process.
+///
+/// Free-threaded CPython 3.14 takes a per-object lock both for `import` and
+/// for module attribute lookup, so resolving this on every `__reduce_ex__`
+/// call makes pickling a contention point instead of a refcount bump.
+/// Measured on 2026-08-07 in a long-lived multi-threaded runtime embedding
+/// this extension: 948 threads parked on the import lock
+/// (_PyMutex_LockTimed → _PyParkingLot_Park → __psynch_cvwait), reached from
+/// PyImport_ImportModuleLevelObject → import_ensure_initialized and from
+/// _Py_module_getattro_impl, with doeff_vm as the caller; the machine paged
+/// out 19.5 GiB. The cell holds the resolved callable so the hot path never
+/// touches the import machinery again.
+static COPYREG_NEWOBJ: PyOnceLock<Py<PyAny>> = PyOnceLock::new();
+
 #[pymethods]
 impl PyEffectBase {
     #[new]
@@ -48,10 +63,14 @@ impl PyEffectBase {
     /// Pickle support: return (copyreg.__newobj__, (cls,), __dict__).
     /// copyreg.__newobj__(cls) calls cls.__new__(cls), then pickle sets
     /// obj.__dict__.update(state) for the third element.
+    ///
+    /// `copyreg.__newobj__` comes from COPYREG_NEWOBJ — resolved once per
+    /// process. Do NOT import or read a module attribute here: this runs on
+    /// every pickle of every effect, and on free-threaded builds those take
+    /// locks (see the COPYREG_NEWOBJ doc comment for the measured wedge).
     fn __reduce_ex__(slf: &Bound<'_, Self>, _protocol: i32) -> PyResult<Py<PyAny>> {
         let py = slf.py();
-        let copyreg = py.import("copyreg")?;
-        let newobj = copyreg.getattr("__newobj__")?;
+        let newobj = COPYREG_NEWOBJ.import(py, "copyreg", "__newobj__")?;
         let cls = slf.get_type();
         let args = pyo3::types::PyTuple::new(py, &[cls.as_any()])?;
         let state = slf.getattr("__dict__")?;

tests/semgrep/fixtures/rust/packages/doeff-vm/src/python_generator_stream.rs

@@ -0,0 +1,9 @@
+// Bad fixture for doeff-vm-no-per-call-copyreg-resolution: the pre-2026-08-10
+// shape of EffectBase.__reduce_ex__, which resolved copyreg.__newobj__ on
+// every call and wedged free-threaded runtimes on the import lock.
+pub fn reduce_ex_bad(slf: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
+    let py = slf.py();
+    let copyreg = py.import("copyreg")?;
+    let newobj = copyreg.getattr("__newobj__")?;
+    Ok(newobj.unbind())
+}

tests/semgrep/test_vm_failfast_semgrep_rules.py

@@ -292,3 +292,31 @@ def test_koine_interactive_terminalize_rule_is_clean_on_fixed_policy() -> None:
 # M3 で消滅し、include が crate src のみの死に rule になるため。readiness
 # discard の不変量は現役側の同契約(sessionhost/launch.hy)の deftest 群が
 # 引き続き守る。rollback 座標 = git tag agentd-rust-final。
+
+
+def test_copyreg_per_call_resolution_rule_detects_pre_fix_reduce_ex() -> None:
+    """doeff-vm-no-per-call-copyreg-resolution は改修前の __reduce_ex__ 形に発火する。
+
+    2026-08-07 の hypha 常駐 runtime 滞留(948 threads が import 鍵で park)の
+    発生源だった「呼び出しごとの copyreg 解決」の回帰ガード。
+    """
+    fixture_root = REPO_ROOT / "tests/semgrep/fixtures/rust"
+    results = _semgrep_results(
+        REPO_ROOT / ".semgrep.yaml",
+        "packages/doeff-vm/src/python_generator_stream.rs",
+        cwd=fixture_root,
+    )
+
+    # py.import("copyreg") と getattr("__newobj__") の 2 行に発火する
+    assert _rule_start_lines(results, "doeff-vm-no-per-call-copyreg-resolution") == {6, 7}
+
+
+def test_copyreg_per_call_resolution_rule_is_clean_on_shipped_source() -> None:
+    """出荷中の python_generator_stream.rs には発火しない(PyOnceLock 化済み)。"""
+    results = _semgrep_results(
+        REPO_ROOT / ".semgrep.yaml",
+        "packages/doeff-vm/src/python_generator_stream.rs",
+        cwd=REPO_ROOT,
+    )
+
+    assert _rule_start_lines(results, "doeff-vm-no-per-call-copyreg-resolution") == set()

tests/test_effect_base_reduce_ex_hot_path.py

@@ -0,0 +1,153 @@
+"""EffectBase.__reduce_ex__ の hot path 不変量 — copyreg 解決は 1 回だけ。
+
+free-threaded CPython 3.14 では import と module 属性参照が per-object の鍵を
+取る。`__reduce_ex__` が呼び出しのたびに `copyreg` を import していると、常駐
+runtime(多スレッド)では import 鍵が競合点になる。
+
+実測(2026-08-07・ACP hypha 常駐 runtime): 948 threads が import 鍵で滞留し、
+機体全体で +19.5 GiB の swap 押し出し。滞留スレッドの native stack は
+_PyMutex_LockTimed→_PyParkingLot_Park→__psynch_cvwait、到達元は
+PyImport_ImportModuleLevelObject→import_ensure_initialized と
+_Py_module_getattro_impl、呼び手は doeff_vm.cpython-314t-darwin.so。
+
+不変量: `copyreg.__newobj__` の解決はプロセスで 1 回だけ(PyOnceLock)。
+hot path には import も module 属性参照も残さない。pickle の往復挙動は不変。
+"""
+
+import builtins
+import copyreg
+import pickle
+import re
+import threading
+from pathlib import Path
+
+import cloudpickle
+from doeff_vm import EffectBase
+
+ROOT = Path(__file__).resolve().parents[1]
+STREAM_RS = ROOT / "packages" / "doeff-vm" / "src" / "python_generator_stream.rs"
+
+
+class Ping(EffectBase):
+    def __init__(self, value, tag="t"):
+        self.value = value
+        self.tag = tag
+
+
+def _reduce_ex_source() -> str:
+    """`__reduce_ex__` の本文だけを Rust ソースから切り出す。"""
+    src = STREAM_RS.read_text()
+    match = re.search(r"fn __reduce_ex__.*?\n    \}", src, re.DOTALL)
+    assert match is not None, f"__reduce_ex__ が {STREAM_RS} に見つからない"
+    return match.group(0)
+
+
+# ---------------------------------------------------------------------------
+# 受入条件 (2): pickle 往復の既存挙動が変わらない
+# ---------------------------------------------------------------------------
+
+
+class TestEffectBasePickle:
+    def test_reduce_ex_shape(self):
+        """(copyreg.__newobj__, (cls,), __dict__) の 3 要素を返す。"""
+        eff = Ping(42)
+        reduced = eff.__reduce_ex__(2)
+        assert isinstance(reduced, tuple)
+        assert len(reduced) == 3
+        assert reduced[0] is copyreg.__newobj__
+        assert reduced[1] == (Ping,)
+        assert reduced[2] == {"value": 42, "tag": "t"}
+
+    def test_pickle_roundtrip(self):
+        restored = pickle.loads(pickle.dumps(Ping(42)))
+        assert isinstance(restored, Ping)
+        assert restored.value == 42
+        assert restored.tag == "t"
+
+    def test_pickle_roundtrip_all_protocols(self):
+        for protocol in range(2, pickle.HIGHEST_PROTOCOL + 1):
+            restored = pickle.loads(pickle.dumps(Ping({"k": [1, 2]}), protocol))
+            assert restored.value == {"k": [1, 2]}, f"protocol={protocol}"
+
+    def test_cloudpickle_roundtrip(self):
+        restored = cloudpickle.loads(cloudpickle.dumps(Ping([1, 2, 3], tag="cp")))
+        assert restored.value == [1, 2, 3]
+        assert restored.tag == "cp"
+
+    def test_nested_effect(self):
+        restored = pickle.loads(pickle.dumps(Ping(Ping(7))))
+        assert restored.value.value == 7
+
+
+# ---------------------------------------------------------------------------
+# 受入条件 (1): hot path に import が無い(実行時の証明 + ソース上の証明)
+# ---------------------------------------------------------------------------
+
+
+class TestReduceExHotPathHasNoImport:
+    def test_no_import_machinery_per_call(self, monkeypatch):
+        """warm-up 後の `__reduce_ex__` は import 機構を一切叩かない。
+
+        `py.import()` は PyImport_Import 経由で builtins.__import__ を呼ぶため、
+        `__import__` の呼び出し回数がそのまま import 鍵に触れた回数になる。
+        (pickle.dumps 自体は save_global で __import__ を呼ぶので、ここでは
+         `__reduce_ex__` を直接叩いて hot path だけを測る。)
+        """
+        eff = Ping(1)
+        eff.__reduce_ex__(2)  # warm-up: 1 回だけの解決を済ませる
+
+        seen: list[str] = []
+        real_import = builtins.__import__
+
+        def counting_import(name, *args, **kwargs):
+            seen.append(name)
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "__import__", counting_import)
+        for _ in range(200):
+            eff.__reduce_ex__(2)
+
+        assert seen == [], f"__reduce_ex__ の hot path が import を呼んだ: {seen}"
+
+    def test_concurrent_reduce_ex_is_consistent(self):
+        """多スレッドから同時に叩いても解決結果は同一かつ正しい。"""
+        results: list[object] = []
+        errors: list[BaseException] = []
+        barrier = threading.Barrier(8)
+
+        def worker():
+            try:
+                barrier.wait()
+                for _ in range(200):
+                    results.append(Ping(1).__reduce_ex__(2)[0])
+            except BaseException as exc:  # スレッド内例外を回収して main で assert する
+                errors.append(exc)
+
+        threads = [threading.Thread(target=worker) for _ in range(8)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert errors == []
+        assert len(results) == 8 * 200
+        assert all(r is copyreg.__newobj__ for r in results)
+
+    def test_source_has_no_per_call_import(self):
+        """ソース上の証明: `__reduce_ex__` 本文に import / getattr 解決が無い。"""
+        body = _reduce_ex_source()
+        assert "py.import(" not in body, (
+            "__reduce_ex__ の hot path に per-call import が復活している "
+            "(free-threaded 3.14 の import 鍵滞留の再発 — 2026-08-07 実測)"
+        )
+        assert "COPYREG_NEWOBJ" in body, (
+            "__newobj__ の解決は静的 PyOnceLock (COPYREG_NEWOBJ) 経由で 1 回だけ行うこと"
+        )
+
+    def test_source_caches_newobj_in_once_lock(self):
+        """静的 PyOnceLock に `copyreg.__newobj__` を保持していること。"""
+        src = STREAM_RS.read_text()
+        assert "PyOnceLock" in src, "PyOnceLock による 1 回きりの解決が無い"
+        assert re.search(
+            r"static\s+COPYREG_NEWOBJ\s*:\s*PyOnceLock<Py<PyAny>>", src
+        ), "copyreg.__newobj__ を保持する静的 PyOnceLock が無い"