fix(parser): remove ICP __del__ loguru cleanup that deadlocked pytest exit

InteractiveContextParser.__del__ called loguru.remove() during GC of
multiple parsers, re-entering loguru's handler lock and hanging after all
tests passed. Replace with an explicit close() and restore spawn for
ProcessPoolExecutor in language derivation.

Co-authored-by: Arash Ashrafzadeh <arashmath16@gmail.com>

Author: cursoragent

src/dylan/parser/interactive_context_parser.py

@@ -144,14 +144,27 @@ def _init_icp_log_settings(
         self._log = loguru_logger.bind(icp_id=self._icp_id)
         sync_dylan_stdlib_level_for_icp(log_level)
 
-    def _configure_icp_sinks(self) -> None:
-        """Register loguru sinks filtered to this parser's ``icp_id`` (or remove prior registrations)."""
+    def _remove_icp_log_sinks(self) -> None:
+        """Remove loguru sinks registered for this parser's ``icp_id``."""
         for hid in self._icp_log_handler_ids:
             try:
                 loguru_logger.remove(hid)
             except ValueError:
                 pass
         self._icp_log_handler_ids.clear()
+
+    def close(self) -> None:
+        """Remove per-parser loguru sinks.
+
+        Call when a parser instance is no longer needed. Do not rely on ``__del__`` for cleanup:
+        removing loguru handlers during interpreter shutdown can deadlock on loguru's internal lock
+        when several parsers are collected together (e.g. after pytest).
+        """
+        self._remove_icp_log_sinks()
+
+    def _configure_icp_sinks(self) -> None:
+        """Register loguru sinks filtered to this parser's ``icp_id`` (or remove prior registrations)."""
+        self._remove_icp_log_sinks()
         if self._log_level == "off":
             return
         lu_level = "ERROR" if self._log_level == "error" else "WARNING"
@@ -183,17 +196,6 @@ def icp_only(record: dict) -> bool:
             )
             self._icp_log_handler_ids.append(hid)
 
-    def __del__(self) -> None:
-        """Best-effort removal of this parser's loguru sinks."""
-        try:
-            for hid in list(getattr(self, "_icp_log_handler_ids", ())):
-                try:
-                    loguru_logger.remove(hid)
-                except ValueError:
-                    pass
-        except Exception:
-            pass
-
     def _log_nonoptional_adjust_limit_exceeded(self) -> None:
         """Emit error when the non-optional adjustment loop exceeds its pass bound."""
         self._log.error(

src/dylan/parser/language_derivation.py

@@ -4,6 +4,7 @@
 
 import itertools
 import logging
+import multiprocessing as mp
 import os
 import random
 from collections import defaultdict
@@ -57,6 +58,15 @@ def _layered_run_status(renderable: str) -> AbstractContextManager[Any]:
 _LAYERED_WORKER_PARSER: Any = None
 
 
+def _language_derivation_mp_context() -> mp.context.BaseContext:
+    """Return a ``spawn`` multiprocessing context for language-derivation worker pools.
+
+    Avoids ``fork()`` from a multi-threaded parent (e.g. pytest on Linux), which can deadlock
+    worker processes and trigger deprecation warnings.
+    """
+    return mp.get_context("spawn")
+
+
 @dataclass(frozen=True)
 class LanguageDerivationRecord:
     """Structured outcome from one candidate sentence in language derivation."""
@@ -685,6 +695,7 @@ def emit_completion_success_only(layer_idx: int, record: LanguageDerivationRecor
             if workers > 1:
                 with ProcessPoolExecutor(
                     max_workers=workers,
+                    mp_context=_language_derivation_mp_context(),
                     initializer=_init_layered_language_worker,
                     initargs=(
                         str(Path(grammar_path)),
@@ -1286,6 +1297,7 @@ def run(
             tasks_iter = ((words, speaker, addressee) for words in candidates)
             with ProcessPoolExecutor(
                 max_workers=workers,
+                mp_context=_language_derivation_mp_context(),
                 initializer=_init_language_worker,
                 initargs=(
                     str(Path(grammar_path)),

tests/test_icp_logging.py

@@ -63,6 +63,16 @@ def test_log_level_warning_suppresses_info() -> None:
     assert not any("Parsed" in msg for _, msg in captured)
 
 
+def test_icp_logging_suite_exits_after_multiple_terminal_parsers() -> None:
+    """Regression: two terminal-log parsers must not deadlock loguru sink removal at GC."""
+    parsers = [
+        InteractiveContextParser(FIXTURE, log_level="error", log_output="terminal"),
+        InteractiveContextParser(FIXTURE, log_level="warning", log_output="terminal"),
+    ]
+    for parser in parsers:
+        parser.close()
+
+
 def test_log_output_file_writes_under_log_dir(tmp_path: Path) -> None:
     """``log_output=file`` with *log_dir* creates a log file and does not require a terminal sink."""
     log_dir = tmp_path / "logs"