fix: a research provider cannot outlast the search deadline

toolgate/executors/research.py

@@ -6,6 +6,8 @@
 """
 from __future__ import annotations
 
+import contextvars
+import threading
 import time
 from datetime import UTC, datetime, timedelta
 
@@ -40,6 +42,33 @@
 from toolgate.executors.research_web import _searx, _tavily
 
 
+class _OutOfTime(Exception):
+    pass
+
+
+def _within(seconds: float, provider, *args):
+    """Run one provider, giving up after `seconds`. A provider makes several calls, each with
+    its own timeout, so only a wall clock keeps it inside the search's deadline. A provider
+    still running is left to finish on its own; its calls end at their own timeouts and its
+    answer is dropped."""
+    outcome: dict = {}
+
+    def run():
+        try:
+            outcome["rows"] = provider(*args)
+        except BaseException as exc:  # handed back to the caller below
+            outcome["error"] = exc
+
+    worker = threading.Thread(target=contextvars.copy_context().run, args=(run,), daemon=True)
+    worker.start()
+    worker.join(seconds)
+    if worker.is_alive():
+        raise _OutOfTime
+    if "error" in outcome:
+        raise outcome["error"]
+    return outcome["rows"]
+
+
 def search(query: str, source: str, limit: int, recency_days: int, until: float | None = None) -> dict:
     """One source's results. `until` is a shared deadline (time.monotonic) from a bundle."""
     if source not in SOURCE_DOMAINS:
@@ -86,14 +115,17 @@ def search(query: str, source: str, limit: int, recency_days: int, until: float
             failures.append(f"{provider_name}: shared deadline exhausted")
             break
         try:
-            rows = provider(clean_query, source, limit, recency_days, min(8.0, remaining))
+            rows = _within(remaining, provider, clean_query, source, limit, recency_days, min(8.0, remaining))
             # Sites the owner skips are never sources, whichever provider found them.
             rows = [row for row in rows if not tool_settings.skipped(row.get("url"), skip_sites)]
             completed_providers.append(provider_name)
             if rows:
                 provider_used = provider_name
                 break
             failures.append(f"{provider_name}: no results")
+        except _OutOfTime:
+            failures.append(f"{provider_name}: shared deadline exhausted")
+            break
         # AttributeError: a provider's JSON with a list or null where an object belongs.
         except (ResearchError, httpx.HTTPError, ValueError, KeyError, TypeError, AttributeError) as exc:
             failures.append(f"{provider_name}: {type(exc).__name__}")

toolgate/tests/test_tool_settings.py

@@ -319,6 +319,24 @@ def unavailable(_query, _source, _limit, _recency_days, timeout):
     assert timeouts and all(0 < timeout <= 5 for timeout in timeouts)
 
 
+def test_a_provider_making_many_calls_cannot_outlast_the_search(store, monkeypatch):
+    calls = []
+
+    def slow_provider(_query, _source, _limit, _recency_days, timeout):
+        for _ in range(3):  # each call stays inside its own timeout; together they do not
+            calls.append(timeout)
+            time.sleep(1.0)
+        return [{"url": "https://example.com/late", "title": "late"}]
+
+    for name in ("_reddit_subreddit_feed", "_reddit_subreddit_searx", "_reddit_subreddit_tavily"):
+        monkeypatch.setattr(research, name, slow_provider)
+    started = time.monotonic()
+    with pytest.raises(research.ResearchError, match="shared deadline exhausted"):
+        research.search("subreddit:accounting", "reddit", 5, 14, until=started + 0.5)
+    assert time.monotonic() - started < 2.0
+    assert len(calls) < 3
+
+
 def test_a_research_bundle_stops_when_the_limit_is_reached(store, monkeypatch):
     now = [0.0]
     deadlines = []