diff --git a/mcp_server/server.py b/mcp_server/server.py index 808e2000..8ca2379d 100644 --- a/mcp_server/server.py +++ b/mcp_server/server.py @@ -348,7 +348,24 @@ def main() -> None: logging.getLogger(__name__).exception("HTTP transport failed. Try with --dashboard flag.") raise else: - mcp.run(transport=STDIO_TRANSPORT) + try: + mcp.run(transport=STDIO_TRANSPORT) + finally: + # stdio ends when the client closes stdin: `mcp.run` returns and the + # process is expected to exit. aiosqlite worker threads are non-daemon, + # so without closing them the interpreter parks in + # `threading._shutdown` and the client's parent sees a server that + # never dies. `finally` so an EOF-time exception does not strand the + # same workers. Logged, not suppressed silently, so a broken close + # stays visible; it still cannot replace the original error. + import asyncio + + from shared.connection import connection_manager + + try: + asyncio.run(connection_manager.close_all()) + except Exception: + logging.getLogger(__name__).warning("close_all() failed during shutdown", exc_info=True) def _run_with_dashboard(host: str, port: int) -> None: diff --git a/scripts/dedup_core_twins.py b/scripts/dedup_core_twins.py index 89ef963f..31839386 100755 --- a/scripts/dedup_core_twins.py +++ b/scripts/dedup_core_twins.py @@ -23,11 +23,14 @@ import asyncio import contextlib import json +import logging import os import sys import time from pathlib import Path +logger = logging.getLogger(__name__) + REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) @@ -93,84 +96,92 @@ async def run(apply: bool) -> int: from shared.constants import DB_NAME from shared.write_dispatch import get_write_dispatch - dispatch = get_write_dispatch(cm) - rows = [dict(r) for r in await dispatch.query(DB_NAME, DUP_SQL)] + try: + dispatch = get_write_dispatch(cm) + rows = [dict(r) for r in await dispatch.query(DB_NAME, DUP_SQL)] - keep, drop = pick(rows) - before_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM core_memory") or 0 + keep, drop = pick(rows) + before_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM core_memory") or 0 - print(f"base={DATA_DIR}") - print(f"mode={'APPLY' if apply else 'dry-run'} dup_rows={len(rows)} groups={len(keep)} drop={len(drop)} total={before_total}") - for k, d in zip(keep, drop, strict=False): - print(f" keep {k['entry_id']} imp={k['importance']:.3f} <- drop {d['entry_id']} imp={d['importance']:.3f} key={d['key']}") + print(f"base={DATA_DIR}") + print(f"mode={'APPLY' if apply else 'dry-run'} dup_rows={len(rows)} groups={len(keep)} drop={len(drop)} total={before_total}") + for k, d in zip(keep, drop, strict=False): + print(f" keep {k['entry_id']} imp={k['importance']:.3f} <- drop {d['entry_id']} imp={d['importance']:.3f} key={d['key']}") - if not apply: - await cm.close_all() - return 0 + if not apply: + return 0 - # guard: every doomed key must address exactly one row - for d in drop: - n = await dispatch.scalar( - DB_NAME, - "SELECT COUNT(*) FROM core_memory WHERE layer=? AND user_id=? AND key=?", - (d["layer"], d["user_id"], d["key"]), - ) - if n != 1: - print(f" ABORT: key {d['key']!r} matches {n} rows, expected 1") - await cm.close_all() - return 1 - - ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) - stamp = time.strftime("%Y%m%d-%H%M%S") - path = ARCHIVE_DIR / f"core-twins-{stamp}.jsonl" - with path.open("w", encoding="utf-8") as fh: + # guard: every doomed key must address exactly one row for d in drop: - fh.write(json.dumps(d, ensure_ascii=False, default=str) + "\n") - - am = ArchivedMemories(cm=cm) - await am._init_db() - now = time.time() - done = 0 - for d in drop: - await am.archive( - user_id=d["user_id"], - content=f"{d['key']}={d['value']}", - memory_type=d["memory_kind"] or "fact", - importance=d["importance"], - original_id=d["entry_id"], - reason="duplicate_twin", - ) - with contextlib.suppress(Exception): - from lifecycle.transitions import record_transition - - await record_transition( - cm, d["user_id"], "l4", f"core:{d['entry_id']}", "archived", f"archived:duplicate_twin:{path.name}", "duplicate_twin" + n = await dispatch.scalar( + DB_NAME, + "SELECT COUNT(*) FROM core_memory WHERE layer=? AND user_id=? AND key=?", + (d["layer"], d["user_id"], d["key"]), ) - l4 = CoreMemory(cm=cm, layer=d["layer"]) - ok = await l4.delete(d["user_id"], d["key"], triggered_by="dedup_core_twins") - if ok: - done += 1 - await dispatch.write( - DB_NAME, - "INSERT INTO audit_log (user_id, action, layer, target_id, details, timestamp) VALUES (?,?,?,?,?,?)", - ( - d["user_id"], - "l4_dedup_twin", - d["layer"], - f"core_memory:{d['entry_id']}", - json.dumps({"key": d["key"], "kept": d["entry_id"], "archive": path.name}, ensure_ascii=False), - now, - ), - ) - - # verify - left = [dict(r) for r in await dispatch.query(DB_NAME, DUP_SQL)] - after_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM core_memory") or 0 - print(f"archive: {path}") - print(f"deleted={done}/{len(drop)} dup_left={len(left)} total {before_total}->{after_total}") - print("VERDICT=" + ("CLEAN" if not left and after_total == before_total - done else "LEFTOVER")) - await cm.close_all() - return 0 if not left and after_total == before_total - done else 1 + if n != 1: + print(f" ABORT: key {d['key']!r} matches {n} rows, expected 1") + return 1 + + ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) + stamp = time.strftime("%Y%m%d-%H%M%S") + path = ARCHIVE_DIR / f"core-twins-{stamp}.jsonl" + with path.open("w", encoding="utf-8") as fh: + for d in drop: + fh.write(json.dumps(d, ensure_ascii=False, default=str) + "\n") + + am = ArchivedMemories(cm=cm) + await am._init_db() + now = time.time() + done = 0 + for d in drop: + await am.archive( + user_id=d["user_id"], + content=f"{d['key']}={d['value']}", + memory_type=d["memory_kind"] or "fact", + importance=d["importance"], + original_id=d["entry_id"], + reason="duplicate_twin", + ) + with contextlib.suppress(Exception): + from lifecycle.transitions import record_transition + + await record_transition( + cm, d["user_id"], "l4", f"core:{d['entry_id']}", "archived", f"archived:duplicate_twin:{path.name}", "duplicate_twin" + ) + l4 = CoreMemory(cm=cm, layer=d["layer"]) + ok = await l4.delete(d["user_id"], d["key"], triggered_by="dedup_core_twins") + if ok: + done += 1 + await dispatch.write( + DB_NAME, + "INSERT INTO audit_log (user_id, action, layer, target_id, details, timestamp) VALUES (?,?,?,?,?,?)", + ( + d["user_id"], + "l4_dedup_twin", + d["layer"], + f"core_memory:{d['entry_id']}", + json.dumps({"key": d["key"], "kept": d["entry_id"], "archive": path.name}, ensure_ascii=False), + now, + ), + ) + + # verify + left = [dict(r) for r in await dispatch.query(DB_NAME, DUP_SQL)] + after_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM core_memory") or 0 + print(f"archive: {path}") + print(f"deleted={done}/{len(drop)} dup_left={len(left)} total {before_total}->{after_total}") + print("VERDICT=" + ("CLEAN" if not left and after_total == before_total - done else "LEFTOVER")) + return 0 if not left and after_total == before_total - done else 1 + finally: + # Close on every exit path, not just the happy one: aiosqlite workers are + # non-daemon, so an exception leaving `run()` without this parks the + # interpreter in `threading._shutdown` (exit 124) instead of printing the + # traceback and exiting. Logged, not suppressed silently, so a broken + # close stays visible; it still cannot replace the original error. + try: + await cm.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) def main() -> int: diff --git a/scripts/dedup_episode_twins.py b/scripts/dedup_episode_twins.py index 1fb266ed..207587f5 100755 --- a/scripts/dedup_episode_twins.py +++ b/scripts/dedup_episode_twins.py @@ -31,12 +31,15 @@ import calendar import contextlib import json +import logging import os import sys import time from collections import Counter, defaultdict from pathlib import Path +logger = logging.getLogger(__name__) + REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) @@ -119,145 +122,150 @@ async def run(apply: bool) -> int: from shared.constants import DB_NAME from shared.write_dispatch import get_write_dispatch - dispatch = get_write_dispatch(cm) - rows = [dict(r) for r in await dispatch.query(DB_NAME, SQL)] - - groups: dict[tuple[str, str], list[dict]] = defaultdict(list) - for r in rows: - groups[(r["user_id"], r["summary"])].append(r) - - keep: list[dict] = [] - drop: list[dict] = [] - klass = Counter() - for members in groups.values(): - k, d, name = classify(members) - klass[name] += 1 - keep.extend(k) - drop.extend(d) - - before_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM episodes") or 0 - orphans_sql = ( - "SELECT COUNT(*) FROM epi_edges e WHERE " - "NOT EXISTS (SELECT 1 FROM epi_nodes n WHERE n.node_id=e.source_id) OR " - "NOT EXISTS (SELECT 1 FROM epi_nodes n WHERE n.node_id=e.target_id)" - ) - before_orphans = await dispatch.scalar(DB_NAME, orphans_sql) or 0 - - print(f"base={DATA_DIR}") - print(f"mode={'APPLY' if apply else 'dry-run'} dup_groups={len(groups)} dup_rows={len(rows)} total={before_total}") - print("classes: " + " ".join(f"{k}={v}" for k, v in sorted(klass.items()))) - print(f"drop={len(drop)} keep={len(keep)}") - - byday = Counter(time.strftime("%Y-%m-%d", time.gmtime(d["created_at"])) for d in drop) - print("drop by day (UTC): " + " ".join(f"{d}:{byday[d]}" for d in sorted(byday))) - - if not apply: - for d in drop[:15]: - print( - f" drop {d['episode_id']:>5} {time.strftime('%m-%d %H:%M', time.gmtime(d['created_at']))} " - f"{_raws(d['tags']).pop():<10} {d['summary'][:52]!r}" - ) - if len(drop) > 15: - print(f" ... +{len(drop) - 15} more") - await cm.close_all() - return 0 - - if not drop: - print("nothing to drop") - await cm.close_all() - return 0 - - # guard: every doomed id must still match its group exactly - doomed = {d["episode_id"] for d in drop} - if len(doomed) != len(drop): - print("ABORT: duplicate episode_id in drop set") - await cm.close_all() - return 1 - for d in drop: - n = await dispatch.scalar( - DB_NAME, - "SELECT COUNT(*) FROM episodes WHERE episode_id=? AND summary=? AND tags IS ? AND layer=?", - (d["episode_id"], d["summary"], d["tags"], d["layer"]), + try: + dispatch = get_write_dispatch(cm) + rows = [dict(r) for r in await dispatch.query(DB_NAME, SQL)] + + groups: dict[tuple[str, str], list[dict]] = defaultdict(list) + for r in rows: + groups[(r["user_id"], r["summary"])].append(r) + + keep: list[dict] = [] + drop: list[dict] = [] + klass = Counter() + for members in groups.values(): + k, d, name = classify(members) + klass[name] += 1 + keep.extend(k) + drop.extend(d) + + before_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM episodes") or 0 + orphans_sql = ( + "SELECT COUNT(*) FROM epi_edges e WHERE " + "NOT EXISTS (SELECT 1 FROM epi_nodes n WHERE n.node_id=e.source_id) OR " + "NOT EXISTS (SELECT 1 FROM epi_nodes n WHERE n.node_id=e.target_id)" ) - if n != 1: - print(f" ABORT: episode {d['episode_id']} no longer matches its group (matches={n})") - await cm.close_all() + before_orphans = await dispatch.scalar(DB_NAME, orphans_sql) or 0 + + print(f"base={DATA_DIR}") + print(f"mode={'APPLY' if apply else 'dry-run'} dup_groups={len(groups)} dup_rows={len(rows)} total={before_total}") + print("classes: " + " ".join(f"{k}={v}" for k, v in sorted(klass.items()))) + print(f"drop={len(drop)} keep={len(keep)}") + + byday = Counter(time.strftime("%Y-%m-%d", time.gmtime(d["created_at"])) for d in drop) + print("drop by day (UTC): " + " ".join(f"{d}:{byday[d]}" for d in sorted(byday))) + + if not apply: + for d in drop[:15]: + print( + f" drop {d['episode_id']:>5} {time.strftime('%m-%d %H:%M', time.gmtime(d['created_at']))} " + f"{_raws(d['tags']).pop():<10} {d['summary'][:52]!r}" + ) + if len(drop) > 15: + print(f" ... +{len(drop) - 15} more") + return 0 + + if not drop: + print("nothing to drop") + return 0 + + # guard: every doomed id must still match its group exactly + doomed = {d["episode_id"] for d in drop} + if len(doomed) != len(drop): + print("ABORT: duplicate episode_id in drop set") return 1 - for k in keep: - if k["episode_id"] in doomed: - print(f" ABORT: keep id {k['episode_id']} also in drop set") - await cm.close_all() - return 1 - - ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) - stamp = time.strftime("%Y%m%d-%H%M%S") - path = ARCHIVE_DIR / f"episode-twins-{stamp}.jsonl" - with path.open("w", encoding="utf-8") as fh: for d in drop: - fh.write(json.dumps(d, ensure_ascii=False, default=str) + "\n") - - am = ArchivedMemories(cm=cm) - await am._init_db() - now = time.time() - for d in drop: - await am.archive( - user_id=d["user_id"], - content=str(d["summary"]), - memory_type="episode", - importance=float(d["emotional_weight"] or 0.0), - original_id=int(d["episode_id"]), - reason="duplicate_episode_replay", - ) - with contextlib.suppress(Exception): - from lifecycle.transitions import record_transition - - await record_transition( - cm, - d["user_id"], - "episode", - f"episode:{d['episode_id']}", - "archived", - f"archived:duplicate_episode_replay:{path.name}", - "duplicate_episode_replay", + n = await dispatch.scalar( + DB_NAME, + "SELECT COUNT(*) FROM episodes WHERE episode_id=? AND summary=? AND tags IS ? AND layer=?", + (d["episode_id"], d["summary"], d["tags"], d["layer"]), ) - await dispatch.write( - DB_NAME, - "INSERT INTO audit_log (user_id, action, layer, target_id, details, timestamp) VALUES (?,?,?,?,?,?)", - ( - d["user_id"], - "l3_dedup_twin", - d["layer"], - f"episodes:{d['episode_id']}", - json.dumps({"raw": sorted(_raws(d["tags"])), "archive": path.name}, ensure_ascii=False), - now, - ), - ) - # API-only delete, each row inside its own (user_id, layer) scope - by_scope: dict[tuple[str, str], list[int]] = defaultdict(list) - for row in drop: - by_scope[(row["user_id"], str(row["layer"]))].append(int(row["episode_id"])) - deleted = 0 - for (owner, layer), ids in by_scope.items(): - deleted += await EpisodicMemory(cm=cm, layer=layer).delete_by_ids(sorted(ids), user_id=owner) - - after_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM episodes") or 0 - after_orphans = await dispatch.scalar(DB_NAME, orphans_sql) or 0 - left = 0 - rest = [dict(r) for r in await dispatch.query(DB_NAME, SQL)] - rg: dict[tuple[str, str], list[dict]] = defaultdict(list) - for r in rest: - rg[(r["user_id"], r["summary"])].append(r) - for members in rg.values(): - if classify(members)[2] == "SAME_RAW": - left += 1 - - print(f"archive: {path}") - print(f"deleted={deleted}/{len(drop)} same_raw_left={left} total {before_total}->{after_total}") - print(f"orphans {before_orphans}->{after_orphans}") - ok = deleted == len(drop) and left == 0 and after_total == before_total - deleted and after_orphans == before_orphans - print("VERDICT=" + ("CLEAN" if ok else "LEFTOVER")) - await cm.close_all() - return 0 if ok else 1 + if n != 1: + print(f" ABORT: episode {d['episode_id']} no longer matches its group (matches={n})") + return 1 + for k in keep: + if k["episode_id"] in doomed: + print(f" ABORT: keep id {k['episode_id']} also in drop set") + return 1 + + ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) + stamp = time.strftime("%Y%m%d-%H%M%S") + path = ARCHIVE_DIR / f"episode-twins-{stamp}.jsonl" + with path.open("w", encoding="utf-8") as fh: + for d in drop: + fh.write(json.dumps(d, ensure_ascii=False, default=str) + "\n") + + am = ArchivedMemories(cm=cm) + await am._init_db() + now = time.time() + for d in drop: + await am.archive( + user_id=d["user_id"], + content=str(d["summary"]), + memory_type="episode", + importance=float(d["emotional_weight"] or 0.0), + original_id=int(d["episode_id"]), + reason="duplicate_episode_replay", + ) + with contextlib.suppress(Exception): + from lifecycle.transitions import record_transition + + await record_transition( + cm, + d["user_id"], + "episode", + f"episode:{d['episode_id']}", + "archived", + f"archived:duplicate_episode_replay:{path.name}", + "duplicate_episode_replay", + ) + await dispatch.write( + DB_NAME, + "INSERT INTO audit_log (user_id, action, layer, target_id, details, timestamp) VALUES (?,?,?,?,?,?)", + ( + d["user_id"], + "l3_dedup_twin", + d["layer"], + f"episodes:{d['episode_id']}", + json.dumps({"raw": sorted(_raws(d["tags"])), "archive": path.name}, ensure_ascii=False), + now, + ), + ) + # API-only delete, each row inside its own (user_id, layer) scope + by_scope: dict[tuple[str, str], list[int]] = defaultdict(list) + for row in drop: + by_scope[(row["user_id"], str(row["layer"]))].append(int(row["episode_id"])) + deleted = 0 + for (owner, layer), ids in by_scope.items(): + deleted += await EpisodicMemory(cm=cm, layer=layer).delete_by_ids(sorted(ids), user_id=owner) + + after_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM episodes") or 0 + after_orphans = await dispatch.scalar(DB_NAME, orphans_sql) or 0 + left = 0 + rest = [dict(r) for r in await dispatch.query(DB_NAME, SQL)] + rg: dict[tuple[str, str], list[dict]] = defaultdict(list) + for r in rest: + rg[(r["user_id"], r["summary"])].append(r) + for members in rg.values(): + if classify(members)[2] == "SAME_RAW": + left += 1 + + print(f"archive: {path}") + print(f"deleted={deleted}/{len(drop)} same_raw_left={left} total {before_total}->{after_total}") + print(f"orphans {before_orphans}->{after_orphans}") + ok = deleted == len(drop) and left == 0 and after_total == before_total - deleted and after_orphans == before_orphans + print("VERDICT=" + ("CLEAN" if ok else "LEFTOVER")) + return 0 if ok else 1 + finally: + # Close on every exit path, not just the happy one: aiosqlite workers are + # non-daemon, so an exception leaving `run()` without this parks the + # interpreter in `threading._shutdown` (exit 124) instead of printing the + # traceback and exiting. Logged, not suppressed silently, so a broken + # close stays visible; it still cannot replace the original error. + try: + await cm.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) async def backfill_transitions(apply: bool) -> int: @@ -273,52 +281,60 @@ async def backfill_transitions(apply: bool) -> int: from shared.constants import DB_NAME from shared.write_dispatch import get_write_dispatch - dispatch = get_write_dispatch(cm) - files = sorted(ARCHIVE_DIR.glob("episode-twins-*.jsonl")) - if not files: - print("no episode-twins-*.jsonl in archive dir — cannot name the to_ref") - await cm.close_all() - return 1 - path = files[-1] - - rows = await dispatch.query( - DB_NAME, - "SELECT id, user_id, original_id, archived_at FROM archived_memories WHERE archive_reason='duplicate_episode_replay' ORDER BY id", - ) - existing = {str(r["from_ref"]) for r in await dispatch.query(DB_NAME, "SELECT from_ref FROM memory_transitions WHERE kind='episode->archived'")} - - todo = [] - for r in rows: - ref = f"episode:{r['original_id']}" - if ref in existing: - continue - todo.append((r["user_id"], ref, r["archived_at"])) - - print(f"archived rows={len(rows)} already recorded={len(rows) - len(todo)} to insert={len(todo)}") - print(f"to_ref=archived:duplicate_episode_replay:{path.name}") - if not apply: - await cm.close_all() - return 0 + try: + dispatch = get_write_dispatch(cm) + files = sorted(ARCHIVE_DIR.glob("episode-twins-*.jsonl")) + if not files: + print("no episode-twins-*.jsonl in archive dir — cannot name the to_ref") + return 1 + path = files[-1] - # One transaction for the whole replay: a half-written transition log would - # claim episodes were archived that the file never recorded. - ops = [] - for user_id, ref, archived_at in todo: - ts = calendar.timegm(time.strptime(str(archived_at), "%Y-%m-%d %H:%M:%S")) - ops.append( - ( - "INSERT INTO memory_transitions (user_id, kind, from_ref, to_ref, reason, ts) VALUES (?,?,?,?,?,?)", - (user_id, "episode->archived", ref, f"archived:duplicate_episode_replay:{path.name}", "duplicate_episode_replay", ts), - ) + rows = await dispatch.query( + DB_NAME, + "SELECT id, user_id, original_id, archived_at FROM archived_memories WHERE archive_reason='duplicate_episode_replay' ORDER BY id", ) - if ops: - await dispatch.write_many(DB_NAME, ops) - - n = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM memory_transitions WHERE reason='duplicate_episode_replay'") or 0 - print(f"inserted={len(todo)} now_in_table={n}") - print("VERDICT=" + ("CLEAN" if n == len(rows) else "LEFTOVER")) - await cm.close_all() - return 0 if n == len(rows) else 1 + existing = { + str(r["from_ref"]) for r in await dispatch.query(DB_NAME, "SELECT from_ref FROM memory_transitions WHERE kind='episode->archived'") + } + + todo = [] + for r in rows: + ref = f"episode:{r['original_id']}" + if ref in existing: + continue + todo.append((r["user_id"], ref, r["archived_at"])) + + print(f"archived rows={len(rows)} already recorded={len(rows) - len(todo)} to insert={len(todo)}") + print(f"to_ref=archived:duplicate_episode_replay:{path.name}") + if not apply: + return 0 + + # One transaction for the whole replay: a half-written transition log would + # claim episodes were archived that the file never recorded. + ops = [] + for user_id, ref, archived_at in todo: + ts = calendar.timegm(time.strptime(str(archived_at), "%Y-%m-%d %H:%M:%S")) + ops.append( + ( + "INSERT INTO memory_transitions (user_id, kind, from_ref, to_ref, reason, ts) VALUES (?,?,?,?,?,?)", + (user_id, "episode->archived", ref, f"archived:duplicate_episode_replay:{path.name}", "duplicate_episode_replay", ts), + ) + ) + if ops: + await dispatch.write_many(DB_NAME, ops) + + n = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM memory_transitions WHERE reason='duplicate_episode_replay'") or 0 + print(f"inserted={len(todo)} now_in_table={n}") + print("VERDICT=" + ("CLEAN" if n == len(rows) else "LEFTOVER")) + return 0 if n == len(rows) else 1 + finally: + # Same contract as `run()`: the dispatch calls above opened connections, + # so any exit path must close them or the interpreter hangs on the + # non-daemon aiosqlite workers. Logged so a broken close is visible. + try: + await cm.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) def main() -> int: diff --git a/scripts/docs_to_wiki.py b/scripts/docs_to_wiki.py index 00524566..8a2411a2 100755 --- a/scripts/docs_to_wiki.py +++ b/scripts/docs_to_wiki.py @@ -25,12 +25,15 @@ import argparse import asyncio import html +import logging import os import re import subprocess import sys from pathlib import Path +logger = logging.getLogger(__name__) + ARIEL_REPO = str(Path(__file__).resolve().parents[1]) # The interpreter running this script is the one whose environment has the # package importable, so re-invoking it needs no path into any particular venv. @@ -117,20 +120,29 @@ def sync_one(source: Path) -> int: return 0 async def _run() -> dict: - from wiki.manager import WikiManager - - wm = WikiManager(layer="user") - # unique parent dirs only — sync_external scans a dir recursively - dirs = sorted({str(p.parent) for p in pages}) - out = {"imported": 0, "skipped": 0, "errors": 0} - for d in dirs: - res = await wm.sync_external([d]) - for k in out: - out[k] += res.get(k, 0) from shared.connection import connection_manager + from wiki.manager import WikiManager - await connection_manager.close_all() - return out + try: + wm = WikiManager(layer="user") + # unique parent dirs only — sync_external scans a dir recursively + dirs = sorted({str(p.parent) for p in pages}) + out = {"imported": 0, "skipped": 0, "errors": 0} + for d in dirs: + res = await wm.sync_external([d]) + for k in out: + out[k] += res.get(k, 0) + return out + finally: + # Close on every exit path, not just the happy one: aiosqlite workers + # are non-daemon, so an exception escaping without this parks the + # interpreter in `threading._shutdown` (exit 124) instead of printing + # the traceback. Logged, not suppressed silently, so a broken close + # stays visible; it still cannot replace the original error. + try: + await connection_manager.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) res = asyncio.run(_run()) print(f"imported={res['imported']} skipped={res['skipped']} errors={res['errors']}") diff --git a/scripts/mirror_identity.py b/scripts/mirror_identity.py index c69278da..a11aa98a 100755 --- a/scripts/mirror_identity.py +++ b/scripts/mirror_identity.py @@ -20,6 +20,7 @@ import asyncio import hashlib import json +import logging import re import sys import time @@ -28,6 +29,8 @@ from shared.constants import DB_NAME +logger = logging.getLogger(__name__) + LIMIT = 2000 # The hourly memory-drain cron writes transient facts into identity files below @@ -213,25 +216,37 @@ async def run_manifest(manifest_path: Path) -> int: original_base = cm.base_dir current_base: Path | None = None total = 0 - for base, file, kind, layer, user, ns in resolved: - if base != current_base: + try: + for base, file, kind, layer, user, ns in resolved: + if base != current_base: + await cm.close_all() + cm.base_dir = base + current_base = base + if file is None: + print(f"drift: {base.name} {kind} missing file") + total += 1 + continue + drift = await check_drift([file], kind, layer, user, cm, ns) + count = sum(len(keys) for keys in drift.values()) + for cls, keys in drift.items(): + for key in keys: + print(f"drift: {base.name} {file.name} {cls} {key}") + total += count + print(f"manifest: {'clean' if not total else f'{total} drift item(s)'} across {len(resolved)} entries") + return total + finally: + # Close on every exit path, not just the happy one: aiosqlite workers are + # non-daemon, so an exception escaping without this parks the interpreter + # in `threading._shutdown` (exit 124) instead of printing the traceback. + # base_dir is restored here too — it is a process-global that must not be + # left pointing at the last manifest base. Logged, not suppressed + # silently, so a broken close stays visible; it still cannot replace the + # original error. + try: await cm.close_all() - cm.base_dir = base - current_base = base - if file is None: - print(f"drift: {base.name} {kind} missing file") - total += 1 - continue - drift = await check_drift([file], kind, layer, user, cm, ns) - count = sum(len(keys) for keys in drift.values()) - for cls, keys in drift.items(): - for key in keys: - print(f"drift: {base.name} {file.name} {cls} {key}") - total += count - await cm.close_all() - cm.base_dir = original_base - print(f"manifest: {'clean' if not total else f'{total} drift item(s)'} across {len(resolved)} entries") - return total + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) + cm.base_dir = original_base async def migrate_sha_keys(mem: Any, cm: Any, user_id: str, layer: str, dry: bool = False) -> tuple[int, list[str]]: @@ -305,23 +320,31 @@ def main() -> int: from shared.connection import connection_manager async def _run() -> int: - app = AppContext() - mem = app.mm.agent_memory(ns.user) if ns.layer == "agent" else app.mm.user_memory(ns.user) - if ns.check: - drift = await check_drift(files, ns.kind, ns.layer, ns.user, connection_manager, ns=ns.ns) - for class_, keys in drift.items(): - for key in keys: - print(f"{class_}: {key}") - total = sum(len(v) for v in drift.values()) - print(f"check: {'clean' if not total else f'{total} drift item(s)'}") - await connection_manager.close_all() - return total - if ns.migrate_sha_keys: - removed, unmapped = await migrate_sha_keys(mem, connection_manager, ns.user, ns.layer, ns.dry_run) - print(f"migrate{' (dry)' if ns.dry_run else ''}: {removed} legacy row(s), {len(unmapped)} unmapped kept") - total = await mirror(files, ns.kind, ns.layer, ns.user, ns.importance, ns.dry_run, mem, connection_manager, ns=ns.ns) - await connection_manager.close_all() - return total + try: + app = AppContext() + mem = app.mm.agent_memory(ns.user) if ns.layer == "agent" else app.mm.user_memory(ns.user) + if ns.check: + drift = await check_drift(files, ns.kind, ns.layer, ns.user, connection_manager, ns=ns.ns) + for class_, keys in drift.items(): + for key in keys: + print(f"{class_}: {key}") + total = sum(len(v) for v in drift.values()) + print(f"check: {'clean' if not total else f'{total} drift item(s)'}") + return total + if ns.migrate_sha_keys: + removed, unmapped = await migrate_sha_keys(mem, connection_manager, ns.user, ns.layer, ns.dry_run) + print(f"migrate{' (dry)' if ns.dry_run else ''}: {removed} legacy row(s), {len(unmapped)} unmapped kept") + return await mirror(files, ns.kind, ns.layer, ns.user, ns.importance, ns.dry_run, mem, connection_manager, ns=ns.ns) + finally: + # Close on every exit path, not just the happy one: aiosqlite workers + # are non-daemon, so an exception escaping without this parks the + # interpreter in `threading._shutdown` (exit 124) instead of printing + # the traceback. Logged, not suppressed silently, so a broken close + # stays visible; it still cannot replace the original error. + try: + await connection_manager.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) n = asyncio.run(_run()) if ns.check: diff --git a/scripts/purge_foreign_rows.py b/scripts/purge_foreign_rows.py index 473b860b..966d3495 100755 --- a/scripts/purge_foreign_rows.py +++ b/scripts/purge_foreign_rows.py @@ -69,11 +69,14 @@ import argparse import asyncio import json +import logging import sqlite3 import sys from pathlib import Path from typing import Any +logger = logging.getLogger(__name__) + REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT)) @@ -283,23 +286,33 @@ async def main() -> int: print(" РЕЖИМ ОТЧЁТА — ничего не записано. Для удаления: --apply") return 0 - before = _counts(db) - print() - await apply(p, db) - after = _counts(db) - print() - print(" изменение:") - for k in before: - d = after[k] - before[k] - print(f" {k:<12} {before[k]:>6} -> {after[k]:<6} ({d:+d})") - from shared.connection import connection_manager from shared.l0 import verify_chain - broken = await verify_chain() - print(f" L0 цепочка нарушений: {len(broken)}") - await connection_manager.close_all() - return 0 + try: + before = _counts(db) + print() + await apply(p, db) + after = _counts(db) + print() + print(" изменение:") + for k in before: + d = after[k] - before[k] + print(f" {k:<12} {before[k]:>6} -> {after[k]:<6} ({d:+d})") + + broken = await verify_chain() + print(f" L0 цепочка нарушений: {len(broken)}") + return 0 + finally: + # Close on every exit path, not just the happy one: aiosqlite workers are + # non-daemon, so an exception escaping without this parks the interpreter + # in `threading._shutdown` (exit 124) instead of printing the traceback. + # Logged, not suppressed silently, so a broken close stays visible; it + # still cannot replace the original error. + try: + await connection_manager.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) if __name__ == "__main__": diff --git a/scripts/resolve_conflicts.py b/scripts/resolve_conflicts.py index bde06642..1ec5970d 100755 --- a/scripts/resolve_conflicts.py +++ b/scripts/resolve_conflicts.py @@ -17,10 +17,13 @@ import argparse import asyncio +import logging import os import sys from pathlib import Path +logger = logging.getLogger(__name__) + REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) @@ -45,40 +48,50 @@ async def run(apply: bool, limit: int = 0) -> int: from shared.write_dispatch import get_write_dispatch cm = connection_manager - dispatch = get_write_dispatch(cm) - rows = await flagged(dispatch) - if limit: - rows = rows[:limit] - print(f"base={DATA_DIR} flagged={len(rows)} mode={'APPLY' if apply else 'dry-run'}") - - resolver = ConflictResolver(cm=cm) - ok = fail = 0 - for row in rows: - rid, gid, preview = row["id"], row["conflict_group_id"], row["preview"] - if not gid: - print(f" SKIP id={rid}: no conflict_group_id") - fail += 1 - continue - if not apply: - print(f" dry id={rid} gid={gid[:8]} {preview[:50]!r}") - continue - try: - if await resolver.resolve(gid, rid): - ok += 1 - print(f" ok id={rid} gid={gid[:8]}") - else: + try: + dispatch = get_write_dispatch(cm) + rows = await flagged(dispatch) + if limit: + rows = rows[:limit] + print(f"base={DATA_DIR} flagged={len(rows)} mode={'APPLY' if apply else 'dry-run'}") + + resolver = ConflictResolver(cm=cm) + ok = fail = 0 + for row in rows: + rid, gid, preview = row["id"], row["conflict_group_id"], row["preview"] + if not gid: + print(f" SKIP id={rid}: no conflict_group_id") fail += 1 - print(f" FAIL id={rid} gid={gid[:8]} (resolve returned False)") - except Exception as e: # one row must not stop the batch - fail += 1 - print(f" ERR id={rid}: {type(e).__name__}: {e}") - - left = len(await flagged(dispatch)) - print(f"done: resolved={ok} failed={fail} flagged_before={len(rows)} flagged_after={left}") - if apply and not limit: - assert left == len(rows) - ok, f"flag count mismatch: {left} != {len(rows) - ok}" - await cm.close_all() - return 1 if fail else 0 + continue + if not apply: + print(f" dry id={rid} gid={gid[:8]} {preview[:50]!r}") + continue + try: + if await resolver.resolve(gid, rid): + ok += 1 + print(f" ok id={rid} gid={gid[:8]}") + else: + fail += 1 + print(f" FAIL id={rid} gid={gid[:8]} (resolve returned False)") + except Exception as e: # one row must not stop the batch + fail += 1 + print(f" ERR id={rid}: {type(e).__name__}: {e}") + + left = len(await flagged(dispatch)) + print(f"done: resolved={ok} failed={fail} flagged_before={len(rows)} flagged_after={left}") + if apply and not limit: + assert left == len(rows) - ok, f"flag count mismatch: {left} != {len(rows) - ok}" + return 1 if fail else 0 + finally: + # Close on every exit path, including the `assert` above: aiosqlite + # workers are non-daemon, so an escaping exception without this parks the + # interpreter in `threading._shutdown` (exit 124) instead of printing the + # traceback and exiting. Logged, not suppressed silently, so a broken + # close stays visible; it still cannot replace the original error. + try: + await cm.close_all() + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) def main() -> int: diff --git a/scripts/warm_embedding_cache.py b/scripts/warm_embedding_cache.py index 17bde4fa..6ea7cf64 100644 --- a/scripts/warm_embedding_cache.py +++ b/scripts/warm_embedding_cache.py @@ -18,6 +18,7 @@ import time logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s", stream=sys.stdout) +logger = logging.getLogger(__name__) BASE = os.environ["MCP_MEMORY_DATA_DIR"] LAYER = "user" @@ -92,11 +93,21 @@ async def main() -> None: print(f" warmed {done} in {time.monotonic() - started:.1f}s", flush=True) -asyncio.run(main()) - # aiosqlite worker threads are non-daemon, so without closing the connections the # interpreter parks in `threading._shutdown` forever after the work is done — the # script looks hung while it has actually finished. from shared.connection import connection_manager as _cm -asyncio.run(_cm.close_all()) +try: + asyncio.run(main()) +finally: + # `finally`, not a plain statement after the call: when `main()` raises (a + # missing table, a bad env, Ctrl-C) the connections are already open, and + # skipping the close turns a clean traceback+exit into a hang that an + # external timeout kills (measured: exit 124 vs exit 1). Logged, not + # suppressed silently, so a broken close stays visible; it still cannot + # replace the traceback the operator needs. + try: + asyncio.run(_cm.close_all()) + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) diff --git a/scripts/warm_ner_cache.py b/scripts/warm_ner_cache.py index 3afec823..51247d10 100644 --- a/scripts/warm_ner_cache.py +++ b/scripts/warm_ner_cache.py @@ -29,6 +29,7 @@ import time logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s", stream=sys.stdout) +logger = logging.getLogger(__name__) BASE = os.environ["MCP_MEMORY_DATA_DIR"] LAYER = "user" @@ -94,11 +95,21 @@ async def main() -> None: ) -asyncio.run(main()) - # aiosqlite worker threads are non-daemon, so without closing the connections the # interpreter parks in `threading._shutdown` forever after the work is done — the # script looks hung while it has actually finished. from shared.connection import connection_manager as _cm -asyncio.run(_cm.close_all()) +try: + asyncio.run(main()) +finally: + # `finally`, not a plain statement after the call: when `main()` raises (a + # missing table, a bad env, Ctrl-C) the connections are already open, and + # skipping the close turns a clean traceback+exit into a hang that an + # external timeout kills (measured: exit 124 vs exit 1). Logged, not + # suppressed silently, so a broken close stays visible; it still cannot + # replace the traceback the operator needs. + try: + asyncio.run(_cm.close_all()) + except Exception: + logger.warning("close_all() failed during shutdown", exc_info=True) diff --git a/tests/test_mcp_server/test_stdio_closes_connections.py b/tests/test_mcp_server/test_stdio_closes_connections.py new file mode 100644 index 00000000..dfc2d4c1 --- /dev/null +++ b/tests/test_mcp_server/test_stdio_closes_connections.py @@ -0,0 +1,56 @@ +"""stdio transport must close aiosqlite connections after `mcp.run` finishes. + +The stdio path used to call `mcp.run(...)` and nothing else. On stdin EOF +`mcp.run` returns and the process is expected to exit, but the aiosqlite worker +threads are non-daemon — without closing them the interpreter parks in +`threading._shutdown` and the MCP client's parent sees a server that never dies. +""" + +from __future__ import annotations + +import sys + +import pytest + + +@pytest.fixture +def stdio_server(monkeypatch): + import mcp_server.server as srv + from shared.connection import connection_manager + + calls: list[str] = [] + + async def _close() -> None: + calls.append("close") + + monkeypatch.setattr(connection_manager, "close_all", _close) + monkeypatch.setattr(srv, "_setup_logging", lambda: None) + monkeypatch.setattr(srv, "_install_faulthandler", lambda: None) + monkeypatch.setattr(sys, "argv", ["ariel-memory", "--transport", "stdio"]) + return srv, calls + + +def test_stdio_closes_after_run_returns(stdio_server, monkeypatch) -> None: + srv, calls = stdio_server + + class _MCP: + def run(self, **_kwargs) -> None: + calls.append("run") + + monkeypatch.setattr(srv, "mcp", _MCP()) + srv.main() + assert calls == ["run", "close"] + + +def test_stdio_closes_when_run_raises(stdio_server, monkeypatch) -> None: + srv, calls = stdio_server + + class _MCP: + def run(self, **_kwargs) -> None: + calls.append("run") + raise RuntimeError("boom") + + monkeypatch.setattr(srv, "mcp", _MCP()) + with pytest.raises(RuntimeError, match="boom"): + srv.main() + assert calls == ["run", "close"] diff --git a/tests/test_scripts/test_close_all_on_failure.py b/tests/test_scripts/test_close_all_on_failure.py new file mode 100644 index 00000000..ccd7ca96 --- /dev/null +++ b/tests/test_scripts/test_close_all_on_failure.py @@ -0,0 +1,209 @@ +"""Every operator script must close its aiosqlite connections on the failure path. + +aiosqlite worker threads are non-daemon: `threading._shutdown` joins them at +interpreter exit, so a script that only closes *after* its work call hangs +forever once that call raises. Reproduced on the exact module-level shape of +`warm_ner_cache.py` / `warm_embedding_cache.py`: exit 124 (external timeout) +without the guard, exit 1 with it. + +Each test drives a real failure through the script's own entry point and +requires the close to be the last thing that happens — a close that ran only on +the happy path leaves the failing call as the last event. +""" + +from __future__ import annotations + +import asyncio +import importlib +import json +import logging +import os +import subprocess +import sys +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 + + +@pytest.mark.parametrize( + "module_name", + ["scripts.dedup_core_twins", "scripts.dedup_episode_twins", "scripts.resolve_conflicts"], +) +async def test_run_closes_when_the_work_call_raises(module_name: str, close_calls: list[str], monkeypatch) -> None: + module = importlib.import_module(module_name) + + def _boom(_cm): + close_calls.append("fail") + raise RuntimeError("boom") + + monkeypatch.setattr("shared.write_dispatch.get_write_dispatch", _boom) + with pytest.raises(RuntimeError, match="boom"): + await module.run(False) + assert close_calls[-1] == "close" + + +async def test_run_manifest_closes_when_drift_check_raises(tmp_path: Path, close_calls: list[str], monkeypatch) -> None: + from shared.connection import connection_manager + + module = importlib.import_module("scripts.mirror_identity") + target = tmp_path / "AGENT.md" + target.write_text("## Alpha\n- a body long enough to be a chunk.\n", encoding="utf-8") + manifest = tmp_path / "manifest.json" + manifest.write_text( + json.dumps([{"base": str(tmp_path / "base"), "file": str(target), "kind": "rule", "layer": "agent"}]), + encoding="utf-8", + ) + original_base = connection_manager.base_dir + + async def _boom(*_args, **_kwargs): + close_calls.append("fail") + raise RuntimeError("boom") + + monkeypatch.setattr(module, "check_drift", _boom) + with pytest.raises(RuntimeError, match="boom"): + await module.run_manifest(manifest) + assert close_calls[-1] == "close" + assert connection_manager.base_dir == original_base + + +def test_mirror_main_closes_when_mirror_raises(tmp_path: Path, close_calls: list[str], monkeypatch) -> None: + module = importlib.import_module("scripts.mirror_identity") + target = tmp_path / "AGENT.md" + target.write_text("## Alpha\n- a body long enough to be a chunk.\n", encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["mirror_identity.py", "--file", str(target)]) + + class _Memory: + def agent_memory(self, _user: str) -> object: + return object() + + class _App: + mm = _Memory() + + async def _boom(*_args, **_kwargs): + close_calls.append("fail") + raise RuntimeError("boom") + + monkeypatch.setattr("mcp_server.context.AppContext", _App) + monkeypatch.setattr(module, "mirror", _boom) + with pytest.raises(RuntimeError, match="boom"): + module.main() + assert close_calls[-1] == "close" + + +def test_docs_to_wiki_closes_when_sync_raises(tmp_path: Path, close_calls: list[str], monkeypatch) -> None: + module = importlib.import_module("scripts.docs_to_wiki") + page = tmp_path / "page.md" + page.write_text("# page\n", encoding="utf-8") + monkeypatch.setattr(module, "convert_to_ssot", lambda _source, _ssot: [page]) + + 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(tmp_path) + 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" + db.write_bytes(b"") + monkeypatch.setattr(sys, "argv", ["purge_foreign_rows.py", "--apply"]) + monkeypatch.setattr(module, "_db_path", lambda: db) + monkeypatch.setattr(module, "plan", lambda _db: {}) + monkeypatch.setattr(module, "report", lambda _plan, _db: None) + monkeypatch.setattr( + module, + "_counts", + lambda _db: {"episodes": 0, "core_memory": 0, "epi_nodes": 0, "epi_edges": 0, "l0_journal": 0}, + ) + + async def _boom(_plan, _db): + close_calls.append("fail") + raise RuntimeError("boom") + + monkeypatch.setattr(module, "apply", _boom) + with pytest.raises(RuntimeError, match="boom"): + asyncio.run(module.main()) + assert close_calls[-1] == "close" + + +@pytest.mark.parametrize("script", ["warm_ner_cache.py", "warm_embedding_cache.py"]) +def test_module_level_scripts_exit_after_the_work_call_raises(script: str, tmp_path: Path) -> None: + """The real repro: no tables in a fresh dir → main() raises after opening a + connection. Without a close on the failure path the process never exits.""" + data_dir = tmp_path / "base" + data_dir.mkdir() + env = {**os.environ, "MCP_MEMORY_DATA_DIR": str(data_dir)} + proc = subprocess.run( + [sys.executable, str(REPO / "scripts" / script)], + env=env, + capture_output=True, + text=True, + timeout=20, + cwd=REPO, + check=False, + ) + assert proc.returncode == 1, proc.stderr + assert "no such table" in proc.stderr + + +async def test_backfill_transitions_closes_when_dispatch_raises(tmp_path: Path, close_calls: list[str], monkeypatch) -> None: + """`--backfill-transitions` is a tenth reachable path: the dispatch calls + open connections and used to close only after them.""" + module = importlib.import_module("scripts.dedup_episode_twins") + monkeypatch.setattr(module, "ARCHIVE_DIR", tmp_path) + (tmp_path / "episode-twins-20260101-000000.jsonl").write_text("{}\n", encoding="utf-8") + + class _Dispatch: + async def query(self, *_args, **_kwargs): + close_calls.append("fail") + raise RuntimeError("boom") + + monkeypatch.setattr("shared.write_dispatch.get_write_dispatch", lambda _cm: _Dispatch()) + with pytest.raises(RuntimeError, match="boom"): + await module.backfill_transitions(False) + assert close_calls[-1] == "close" + + +def test_close_failure_on_success_path_is_logged(caplog, monkeypatch) -> None: + """A close that raises after successful work must leave a trace, not vanish + behind `contextlib.suppress` — and the return code stays as it was.""" + module = importlib.import_module("scripts.resolve_conflicts") + from shared.connection import connection_manager + + class _Dispatch: + async def query(self, *_args, **_kwargs): + return [] + + async def _boom() -> None: + raise RuntimeError("close boom") + + monkeypatch.setattr("shared.write_dispatch.get_write_dispatch", lambda _cm: _Dispatch()) + monkeypatch.setattr(connection_manager, "close_all", _boom) + with caplog.at_level(logging.WARNING): + rc = asyncio.run(module.run(False)) + assert rc == 0 + assert any("close_all" in record.getMessage() for record in caplog.records), caplog.text