test(web/ddgs): cover GIL-hold timeout, interrupt, and worker reap

Regression tests for #68096: native GIL-hold and sleep hooks must time
out or interrupt promptly with no orphaned search workers.

tests/tools/test_web_providers_ddgs.py

@@ -4,6 +4,7 @@ Covers:
 - DDGSWebSearchProvider.is_available() — reflects package importability
 - DDGSWebSearchProvider.search() — happy path, missing package, runtime error
 - Result normalization (title, url, description, position)
+- Process-isolated timeout / interrupt / GIL-hold / reap (#68096)
 - _is_backend_available("ddgs") / _get_backend() integration
 - web_extract returns a search-only error when ddgs is active
 """
@@ -11,6 +12,7 @@ from __future__ import annotations
 
 import json
 import sys
+import time
 import types
 
 import pytest
@@ -52,6 +54,21 @@ def _install_fake_ddgs(monkeypatch, *, text_results=None, text_raises=None, text
     return fake
 
 
+def _force_inprocess_search(monkeypatch, prov):
+    """Route bounded search through the in-process helper.
+
+    Happy-path unit tests install a fake ``ddgs`` in the parent interpreter;
+    spawn workers would not see that fake. Isolation behavior is covered by
+    dedicated process tests below.
+    """
+    monkeypatch.setattr(
+        prov,
+        "_run_ddgs_search_bounded",
+        lambda query, safe_limit: prov._run_ddgs_search(query, safe_limit),
+        raising=True,
+    )
+
+
 # ---------------------------------------------------------------------------
 # DDGSWebSearchProvider unit tests
 # ---------------------------------------------------------------------------
@@ -98,9 +115,10 @@ class TestDDGSProviderSearch:
             {"title": "B", "href": "https://b.example.com", "body": "desc B"},
             {"title": "C", "href": "https://c.example.com", "body": "desc C"},
         ])
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("q", limit=5)
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
 
         assert result["success"] is True
         web = result["data"]["web"]
@@ -112,9 +130,10 @@ class TestDDGSProviderSearch:
         _install_fake_ddgs(monkeypatch, text_results=[
             {"title": "A", "url": "https://a.example.com", "body": "desc A"},
         ])
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("q", limit=5)
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
 
         assert result["success"] is True
         assert result["data"]["web"][0]["url"] == "https://a.example.com"
@@ -124,9 +143,10 @@ class TestDDGSProviderSearch:
             {"title": f"R{i}", "href": f"https://r{i}.example.com", "body": ""}
             for i in range(10)
         ])
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("q", limit=3)
+        result = prov.DDGSWebSearchProvider().search("q", limit=3)
 
         assert result["success"] is True
         assert len(result["data"]["web"]) == 3
@@ -151,54 +171,42 @@ class TestDDGSProviderSearch:
 
     def test_runtime_error_returns_failure(self, monkeypatch):
         _install_fake_ddgs(monkeypatch, text_raises=RuntimeError("rate limited 202"))
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("q", limit=5)
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
         assert result["success"] is False
         assert "rate limited" in result["error"] or "failed" in result["error"].lower()
 
     def test_empty_results(self, monkeypatch):
         _install_fake_ddgs(monkeypatch, text_results=[])
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("nothing", limit=5)
+        result = prov.DDGSWebSearchProvider().search("nothing", limit=5)
         assert result["success"] is True
         assert result["data"]["web"] == []
 
+    @pytest.mark.live_system_guard_bypass
     def test_hung_search_times_out_and_returns_failure(self, monkeypatch):
-        """#36776: a ddgs call that never returns must be bounded by the
-        wall-clock timeout and surface a failure instead of hanging the
-        shared agent loop. We patch the blocking helper to wait on an Event
-        (released in finally so no worker thread leaks past the test) and
-        shrink the timeout; search() must return success=False promptly."""
-        import threading
-        import time
-
-        # ddgs must import-probe True for search() to proceed.
+        """#36776 / #68096: a hung worker must be bounded by the wall-clock
+        timeout and reaped — even when the child never returns to Python."""
         _install_fake_ddgs(monkeypatch)
-        monkeypatch.delitem(sys.modules, "plugins.web.ddgs.provider", raising=False)
-        import plugins.web.ddgs.provider as _prov
+        import plugins.web.ddgs.provider as prov
 
-        release = threading.Event()
+        monkeypatch.setattr(prov, "_test_hook", "sleep", raising=True)
+        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 0.4, raising=True)
+        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
 
-        def _blocking_search(query, safe_limit):
-            release.wait(timeout=10)  # bounded so the worker can never truly leak
-            return []
+        start = time.monotonic()
+        result = prov.DDGSWebSearchProvider().search("hangs forever", limit=5)
+        elapsed = time.monotonic() - start
 
-        monkeypatch.setattr(_prov, "_run_ddgs_search", _blocking_search, raising=True)
-        monkeypatch.setattr(_prov, "_SEARCH_TIMEOUT_SECS", 0.3, raising=True)
-
-        try:
-            start = time.monotonic()
-            result = _prov.DDGSWebSearchProvider().search("hangs forever", limit=5)
-            elapsed = time.monotonic() - start
-
-            assert result["success"] is False
-            assert "timed out" in result["error"].lower()
-            # Returned well before the worker's 10s wait — proves the cap fired.
-            assert elapsed < 3.0, f"search did not return promptly ({elapsed:.1f}s)"
-        finally:
-            release.set()  # let the orphaned worker finish immediately
+        assert result["success"] is False
+        assert "timed out" in result["error"].lower()
+        assert elapsed < 5.0, f"search did not return promptly ({elapsed:.1f}s)"
+        _assert_worker_reaped(prov)
 
     def test_fast_search_not_affected_by_timeout_wrapper(self, monkeypatch):
         """Happy-path guard: the timeout wrapper must not break a normal,
@@ -207,14 +215,115 @@ class TestDDGSProviderSearch:
             monkeypatch,
             text_results=[{"title": "T", "href": "https://e.com", "body": "B"}],
         )
-        from plugins.web.ddgs.provider import DDGSWebSearchProvider
+        import plugins.web.ddgs.provider as prov
+        _force_inprocess_search(monkeypatch, prov)
 
-        result = DDGSWebSearchProvider().search("q", limit=5)
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
         assert result["success"] is True
         assert result["data"]["web"][0]["url"] == "https://e.com"
         assert result["data"]["web"][0]["title"] == "T"
 
 
+# ---------------------------------------------------------------------------
+# Process isolation (#68096)
+# ---------------------------------------------------------------------------
+
+
+def _assert_worker_reaped(prov) -> None:
+    """Assert the last DDGS worker process has exited."""
+    proc = prov._last_worker_proc
+    assert proc is not None, "expected a DDGS worker process to have been started"
+    assert proc.poll() is not None, (
+        f"DDGS worker still alive (pid={proc.pid}, returncode={proc.returncode})"
+    )
+
+
+@pytest.mark.live_system_guard_bypass
+class TestDDGSProcessIsolation:
+    def test_gil_holding_worker_times_out_and_is_reaped(self, monkeypatch):
+        """#68096: parent deadline still fires when the child holds its GIL."""
+        _install_fake_ddgs(monkeypatch)
+        import plugins.web.ddgs.provider as prov
+
+        monkeypatch.setattr(prov, "_test_hook", "gil", raising=True)
+        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 0.5, raising=True)
+        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
+
+        start = time.monotonic()
+        result = prov.DDGSWebSearchProvider().search("gil hold", limit=5)
+        elapsed = time.monotonic() - start
+
+        assert result["success"] is False
+        assert "timed out" in result["error"].lower()
+        assert elapsed < 5.0, f"GIL-hold search did not time out promptly ({elapsed:.1f}s)"
+        _assert_worker_reaped(prov)
+
+    def test_interrupt_terminates_worker_promptly(self, monkeypatch):
+        """TUI/gateway interrupt must kill the DDGS child before the deadline."""
+        _install_fake_ddgs(monkeypatch)
+        import plugins.web.ddgs.provider as prov
+
+        # Flip interrupt after the first poll so the wait loop observes it.
+        calls = {"n": 0}
+
+        def _interrupt_after_poll():
+            calls["n"] += 1
+            return calls["n"] >= 2
+
+        monkeypatch.setattr(prov, "_test_hook", "sleep", raising=True)
+        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 30, raising=True)
+        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", _interrupt_after_poll)
+
+        start = time.monotonic()
+        result = prov.DDGSWebSearchProvider().search("interrupt me", limit=5)
+        elapsed = time.monotonic() - start
+
+        assert result["success"] is False
+        assert "interrupted" in result["error"].lower()
+        assert elapsed < 5.0, f"interrupt did not return promptly ({elapsed:.1f}s)"
+        _assert_worker_reaped(prov)
+
+    def test_spawned_worker_success_envelope(self, monkeypatch):
+        """Real spawn path: success envelope round-trips through the pipe."""
+        _install_fake_ddgs(monkeypatch)
+        import plugins.web.ddgs.provider as prov
+
+        monkeypatch.setattr(prov, "_test_hook", "success", raising=True)
+        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 5, raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
+
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
+        assert result["success"] is True
+        assert result["data"]["web"][0]["url"] == "https://example.com"
+        _assert_worker_reaped(prov)
+
+    def test_spawned_worker_error_envelope(self, monkeypatch):
+        """Real spawn path: error envelope becomes success=False."""
+        _install_fake_ddgs(monkeypatch)
+        import plugins.web.ddgs.provider as prov
+
+        monkeypatch.setattr(prov, "_test_hook", "error", raising=True)
+        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 5, raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
+
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
+        assert result["success"] is False
+        assert "boom" in result["error"]
+        _assert_worker_reaped(prov)
+
+    def test_no_orphan_after_successful_search(self, monkeypatch):
+        _install_fake_ddgs(monkeypatch)
+        import plugins.web.ddgs.provider as prov
+
+        monkeypatch.setattr(prov, "_test_hook", "empty", raising=True)
+        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
+
+        result = prov.DDGSWebSearchProvider().search("q", limit=5)
+        assert result["success"] is True
+        _assert_worker_reaped(prov)
+
 # ---------------------------------------------------------------------------
 # Integration: _is_backend_available / _get_backend / check_web_api_key
 # ---------------------------------------------------------------------------