fix(scripts): the last three close sites, and stop the test pollution I shipped

scripts/backfill_memory_kind.py

@@ -18,7 +18,6 @@
 
 import argparse
 import asyncio
-import contextlib
 import logging
 from typing import Any
 
@@ -74,8 +73,13 @@ async def run() -> int:
                 limit=args.limit,
             )
         finally:
-            with contextlib.suppress(Exception):
+            # A close failure is the only signal that aiosqlite workers may still
+            # be alive, so suppressing it hides the very hang this guard exists
+            # to prevent. Logged, not raised: it must not replace the work error.
+            try:
                 await connection_manager.close_all()
+            except Exception:
+                logger.warning("close_all() failed during shutdown", exc_info=True)
         print(f"{'applied' if args.apply else 'dry-run'}: {n} chunk(s)")
         return 0
 

scripts/sync_skills.py

@@ -20,8 +20,9 @@
 
 import argparse
 import asyncio
-import shutil
+import logging
 import os
+import shutil
 import subprocess
 import sys
 from pathlib import Path
@@ -37,6 +38,8 @@
     "cowagent": Path.home() / ".mcp-ariel-memory-cowagent",
 }
 
+logger = logging.getLogger(__name__)
+
 README = """# Skills SSOT
 
 Shared skill set for all agents (D2.3). Every `.md` under `skill/` is synced
@@ -69,13 +72,20 @@ def sync_one(agent: str) -> dict:
 
     async def _run() -> dict:
         wm = WikiManager(layer="user")
-        res = await wm.sync_external([str(SSOT_DIR / "skill")])
-        # aiosqlite workers are non-daemon: without close_all the interpreter
-        # hangs at exit AFTER printing (autohooks _close_ariel lesson).
-        from shared.connection import connection_manager
-
-        await connection_manager.close_all()
-        return res
+        try:
+            return await wm.sync_external([str(SSOT_DIR / "skill")])
+        finally:
+            # Close on every exit path: aiosqlite workers are non-daemon, so an
+            # exception escaping sync_external without this parks the interpreter
+            # in threading._shutdown (the per-agent subprocess hangs instead of
+            # failing). Logged, not suppressed silently, so a broken close stays
+            # visible; it still cannot replace the original error.
+            from shared.connection import connection_manager
+
+            try:
+                await connection_manager.close_all()
+            except Exception:
+                logger.warning("close_all() failed during shutdown", exc_info=True)
 
     return asyncio.run(_run())
 

scripts/verify_autohooks.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import sqlite3
 import sys
@@ -30,6 +31,8 @@
 
 RESULTS: list[tuple[str, bool, str]] = []
 
+logger = logging.getLogger(__name__)
+
 
 def check(name: str, ok: bool, detail: str = "") -> None:
     RESULTS.append((name, ok, detail))
@@ -70,7 +73,7 @@ def make_schemas() -> None:
     conv.close()
 
 
-async def main() -> int:
+async def _run_checks() -> list[tuple[str, bool, str]]:
     make_schemas()
     from mcp_server.context import AppContext
 
@@ -278,15 +281,32 @@ async def _capturing_dispatch(event, layer, user_id, payload, m, g, r=None):
         check("HTTP surfaces", False, repr(e)[:150])
 
     # ── summary ───────────────────────────────────────────────────────────
-    from shared.connection import connection_manager
-
-    await connection_manager.close_all()  # aiosqlite workers block interpreter exit otherwise
     failed = [r for r in RESULTS if not r[1]]
     print(
         f"\n{'=' * 64}\n{len(RESULTS) - len(failed)}/{len(RESULTS)} checks passed"
         + (f" — FAILURES: {[f[0] for f in failed]}" if failed else " — ALL GREEN")
     )
     sys.stdout.flush()
+    return failed
+
+
+async def main() -> int:
+    try:
+        failed = await _run_checks()
+    finally:
+        # Close on every exit path, not just the happy one: aiosqlite workers are
+        # non-daemon, so an exception escaping the checks leaves the interpreter
+        # parked in threading._shutdown until the 45 s faulthandler fires and
+        # replaces the real failure with a timeout exit. Logged, not suppressed
+        # silently, so a broken close stays visible; it still cannot replace the
+        # original error. `os._exit` below would skip a `finally` it sat inside,
+        # which is why the close lives here and the exit does not.
+        from shared.connection import connection_manager
+
+        try:
+            await connection_manager.close_all()
+        except Exception:
+            logger.warning("close_all() failed during shutdown", exc_info=True)
     # Known repo-wide aiosqlite quirk: a lingering worker thread blocks interpreter
     # shutdown (same workaround as tests/conftest.py pytest_sessionfinish).
     os._exit(1 if failed else 0)

tests/test_scripts/test_backfill_memory_kind.py

@@ -6,6 +6,9 @@
 
 from __future__ import annotations
 
+import importlib
+import logging
+
 import pytest
 
 from shared.connection import connection_manager
@@ -77,3 +80,23 @@ async def test_backfill_limit(db) -> None:
     await _chunk("нужно сделать два")
     assert await backfill_memory_kind(connection_manager, dry_run=False, limit=1) == 1
     assert await backfill_memory_kind(connection_manager, dry_run=False, limit=1) == 1
+
+
+def test_close_failure_is_logged_not_suppressed(caplog, monkeypatch) -> None:
+    """The `finally` was already there; the defect was `contextlib.suppress`,
+    which made a real close failure indistinguishable from a clean shutdown.
+    A broken close must leave a warning and must not change the exit code."""
+    module = importlib.import_module("scripts.backfill_memory_kind")
+
+    async def _work(*_args, **_kwargs) -> int:
+        return 0
+
+    async def _boom() -> None:
+        raise RuntimeError("close boom")
+
+    monkeypatch.setattr(module, "backfill_memory_kind", _work)
+    monkeypatch.setattr(connection_manager, "close_all", _boom)
+    with caplog.at_level(logging.WARNING):
+        rc = module.main([])
+    assert rc == 0
+    assert any("close_all" in record.getMessage() for record in caplog.records), caplog.text

tests/test_scripts/test_close_all_on_failure.py

@@ -21,12 +21,35 @@
 import subprocess
 import sys
 from pathlib import Path
+from types import ModuleType
 
 import pytest
 
 REPO = Path(__file__).resolve().parents[2]
 
 
+@pytest.fixture(autouse=True)
+def _rebind_leaked_write_dispatch():
+    """Rebind `get_write_dispatch` in modules that imported it during a patch.
+
+    `core.memory`, `core.episodic`, `core.session` and ~40 others bind the
+    function object at module level. A test that patches
+    `shared.write_dispatch.get_write_dispatch` and then imports one of them
+    inside that window leaves the stub permanently bound: monkeypatch restores
+    the shared attribute, never the copies already taken. Rebinding at teardown
+    keeps the stub local to the test that asked for it.
+    """
+    from shared import write_dispatch
+
+    real = write_dispatch.get_write_dispatch
+    yield
+    for module in list(sys.modules.values()):
+        if not isinstance(module, ModuleType):
+            continue
+        if getattr(module, "get_write_dispatch", real) is not real:
+            module.get_write_dispatch = real  # type: ignore[attr-defined]
+
+
 @pytest.fixture
 def close_calls(monkeypatch) -> list[str]:
     """Record close_all() on the shared singleton, leaving the real one unused."""
@@ -126,6 +149,25 @@ async def sync_external(self, _dirs) -> dict[str, int]:
     assert close_calls[-1] == "close"
 
 
+def test_sync_skills_closes_when_sync_raises(close_calls: list[str], monkeypatch) -> None:
+    """`sync_one` runs `sync_external` before closing, so a raise there used to
+    strand the aiosqlite workers and hang the per-agent subprocess."""
+    module = importlib.import_module("scripts.sync_skills")
+
+    class _WikiManager:
+        def __init__(self, **_kwargs) -> None:
+            pass
+
+        async def sync_external(self, _dirs) -> dict[str, int]:
+            close_calls.append("fail")
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr("wiki.manager.WikiManager", _WikiManager)
+    with pytest.raises(RuntimeError, match="boom"):
+        module.sync_one("pytest")
+    assert close_calls[-1] == "close"
+
+
 def test_purge_foreign_rows_closes_when_apply_raises(tmp_path: Path, close_calls: list[str], monkeypatch) -> None:
     module = importlib.import_module("scripts.purge_foreign_rows")
     db = tmp_path / "memory.db"

tests/test_scripts/test_verify_autohooks_close.py

@@ -0,0 +1,105 @@
+"""`verify_autohooks.py` must close its connections when the body raises.
+
+The script ends with `os._exit` to sidestep the repo-wide aiosqlite teardown
+quirk, and arms `faulthandler.dump_traceback_later(45, exit=True)`. Both are
+deliberate. The defect was that `close_all` sat on the success path only: an
+exception between the work and the end left the non-daemon aiosqlite workers
+alive, so the interpreter parked in `threading._shutdown` until the 45 s guard
+fired and the exit code came from faulthandler instead of the real failure.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+
+import pytest
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+@pytest.fixture
+def close_calls(monkeypatch) -> list[str]:
+    """Record close_all() on the shared singleton, leaving the real one unused."""
+    from shared.connection import connection_manager
+
+    calls: list[str] = []
+
+    async def _close() -> None:
+        calls.append("close")
+
+    monkeypatch.setattr(connection_manager, "close_all", _close)
+    return calls
+
+
+def test_main_closes_when_the_body_raises(close_calls: list[str], monkeypatch) -> None:
+    module = importlib.import_module("scripts.verify_autohooks")
+
+    def _boom() -> None:
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(module, "make_schemas", _boom)
+    with pytest.raises(RuntimeError, match="boom"):
+        asyncio.run(module.main())
+    assert close_calls[-1] == "close"
+
+
+def test_a_raising_body_exits_promptly_instead_of_waiting_for_the_45s_guard(tmp_path: Path) -> None:
+    """Proof of the hang: an async seam opens a real connection, then raises.
+    With a close in `finally` the process prints the traceback and exits in
+    seconds; without one it is still alive at 20 s, waiting for the 45 s
+    faulthandler."""
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    code = textwrap.dedent(
+        f"""
+        import asyncio, faulthandler, sys, time
+        sys.path.insert(0, {str(REPO)!r})
+        import scripts.verify_autohooks as v
+        import hooks.external as hx
+
+        async def _boom(*_args, **_kwargs):
+            from shared.connection import connection_manager
+            from shared.constants import DB_NAME
+            await connection_manager.get(DB_NAME)
+            raise RuntimeError("boom")
+
+        hx.dispatch_event = _boom
+        faulthandler.dump_traceback_later(45, exit=True)
+        started = time.monotonic()
+        try:
+            asyncio.run(v.main())
+        finally:
+            print(f"ELAPSED={{time.monotonic() - started:.2f}}")
+        """
+    )
+
+    env = {
+        **os.environ,
+        "MCP_MEMORY_DATA_DIR": str(data_dir),
+        "ARIEL_HASH_EMBEDDINGS": "1",
+        "BACKUP_CRON_DISABLED": "1",
+    }
+    try:
+        proc = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd=REPO,
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=20,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        pytest.fail("the script did not exit within 20s — the 45s faulthandler guard was still in charge")
+
+    assert proc.returncode != 0, proc.stdout
+    assert "RuntimeError: boom" in proc.stderr, proc.stderr
+    assert "Timeout (0:00:45)" not in proc.stderr, proc.stderr
+    elapsed = float(proc.stdout.split("ELAPSED=")[-1].split()[0])
+    assert elapsed < 10, proc.stdout