diff --git a/scripts/backfill_memory_kind.py b/scripts/backfill_memory_kind.py index 6fbab06..3693123 100644 --- a/scripts/backfill_memory_kind.py +++ b/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 diff --git a/scripts/sync_skills.py b/scripts/sync_skills.py index 6b34f8c..bc5ac8e 100755 --- a/scripts/sync_skills.py +++ b/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()) diff --git a/scripts/verify_autohooks.py b/scripts/verify_autohooks.py index e46a84e..d0c8f82 100755 --- a/scripts/verify_autohooks.py +++ b/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) diff --git a/tests/test_scripts/test_backfill_memory_kind.py b/tests/test_scripts/test_backfill_memory_kind.py index a566a8d..ba73306 100644 --- a/tests/test_scripts/test_backfill_memory_kind.py +++ b/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 diff --git a/tests/test_scripts/test_close_all_on_failure.py b/tests/test_scripts/test_close_all_on_failure.py index ccd7ca9..193ba56 100644 --- a/tests/test_scripts/test_close_all_on_failure.py +++ b/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" diff --git a/tests/test_scripts/test_verify_autohooks_close.py b/tests/test_scripts/test_verify_autohooks_close.py new file mode 100644 index 0000000..468f7df --- /dev/null +++ b/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