From 5313479b4ecb9b63b3e6d41a78a1f5ee3b38c4f0 Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sun, 18 Oct 2026 07:14:26 +0000 Subject: [PATCH 1/4] Add anonymous Parallel MCP search to OfficeQA rollouts --- CHANGELOG.md | 4 + configs/officeqa/default.yaml | 7 +- docs/reference/config.md | 32 +++ pyproject.toml | 1 + skillopt/envs/officeqa/adapter.py | 12 +- skillopt/envs/officeqa/parallel_search.py | 96 +++++++++ skillopt/envs/officeqa/rollout.py | 97 ++++++--- tests/test_officeqa_parallel_search.py | 237 ++++++++++++++++++++++ 8 files changed, 450 insertions(+), 36 deletions(-) create mode 100644 skillopt/envs/officeqa/parallel_search.py create mode 100644 tests/test_officeqa_parallel_search.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c61a74dd6..5ee77088f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ All notable changes to SkillOpt are documented here. This project adheres to ## [Unreleased] ### Added +- **OfficeQA Parallel Search MCP**: fresh `auto` search configurations use + anonymous Fast-mode web search with `openai_chat`/`qwen_chat` when no custom + service is configured. Saved offline, custom-service and Azure search modes + remain explicit choices; other harnesses retain their local tool permissions. - **SkillOpt-Sleep paired A/B evalkit** (`python -m skillopt_sleep.evalkit`): McNemar plus percentile-bootstrap CIs on a fixed task manifest, with task-cluster multi-seed inference and seeded null calibration for exact-test diff --git a/configs/officeqa/default.yaml b/configs/officeqa/default.yaml index 7b72f1a4b..e4c2be458 100644 --- a/configs/officeqa/default.yaml +++ b/configs/officeqa/default.yaml @@ -24,11 +24,12 @@ env: workers: 4 max_tool_turns: 24 max_completion_tokens: 16384 - search_mode: offline + # Fresh chat runs use keyless Parallel; saved offline/custom/Azure modes remain explicit. + search_mode: auto max_queries_per_turn: 4 - search_api_url: http://apisix.westus2.cloudapp.azure.com/search_tool/search + search_api_url: null search_auth_env: OFFICEQA_CUSTOM_SEARCH_AUTH - search_provider: duckduckgo + search_provider: null search_max_num_results: 4 search_timeout_seconds: 20 limit: 0 diff --git a/docs/reference/config.md b/docs/reference/config.md index 6371f69dd..177784a7e 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -210,6 +210,38 @@ forwarded consistently to both SDK and CLI execution paths. Benchmark-specific `env` keys are passed through to the adapter. +### OfficeQA web search + +New OfficeQA configs use `env.search_mode: auto`. With `openai_chat` or +`qwen_chat`, an unconfigured search service uses +[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) +at `https://search.parallel.ai/mcp`. Search runs in Fast mode without an +account or API key. The controller sends the model's `` to +MCP and supplies the returned URLs and excerpts to the next model round. +Parallel search does not require a local OfficeQA document corpus; parsed +source-page context is included when that corpus is available. Model backend +configuration is still required. + +| Setting | Behavior | +|---|---| +| `env.search_mode: auto` | Use a configured custom service when `search_api_url`, `search_provider`, `OFFICEQA_SEARCH_API_URL`, or the token named by `search_auth_env` is set; otherwise use Parallel for `openai_chat`/`qwen_chat`. Other backends, or `use_local_tools: false`, retain offline mode. | +| `env.search_mode: parallel` | Explicitly select anonymous Parallel search with `openai_chat`/`qwen_chat`, ignoring custom service settings. | +| `env.search_mode: offline` | Keep search local to the configured document corpus, with no Parallel connection. `env.use_local_tools: false` also disables local function tools in this mode. | +| `env.search_mode: custom_search` | Keep the existing token-authenticated HTTP service and provider. Set `search_api_url`, `search_auth_env` and `search_provider` for your service. | +| `env.search_mode: azure_search` | Keep the existing model-provided Azure search path. | +| `env.search_timeout_seconds` | Whole MCP discovery/search deadline in seconds (default `20`). | +| `env.search_max_num_results` | Maximum number of Parallel evidence results per round (default `4`). | +| `env.max_queries_per_turn` / `env.max_tool_turns` | Bound model-requested queries and model rounds as before. | + +Existing saved `offline`, `custom_search`, and `azure_search` settings remain +authoritative. To keep a new experiment local, set `env.search_mode=offline` +with `--cfg-options`. To switch a saved config to Parallel, set +`env.search_mode=parallel`. Explicit providers never fall back to Parallel on +failure. Anonymous requests do not use `PARALLEL_API_KEY` or saved MCP +credentials; service errors, including free-tier limits, fail the rollout +with a search diagnostic. This provider changes the evidence source, so its +scores are not a reproduction of the paper's Google-backed search setting. + ## Credential Environment Variables ### Azure-family backend diff --git a/pyproject.toml b/pyproject.toml index 5d50b8fef..c3cd9b450 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ "azure-identity>=1.15.0", "azure-core>=1.30.0", "httpx>=0.27.0", + "mcp>=1.27.0,<2", ] [project.optional-dependencies] diff --git a/skillopt/envs/officeqa/adapter.py b/skillopt/envs/officeqa/adapter.py index 63419d48f..37ea47b15 100644 --- a/skillopt/envs/officeqa/adapter.py +++ b/skillopt/envs/officeqa/adapter.py @@ -1,7 +1,5 @@ from __future__ import annotations -import os - from skillopt.datasets.base import BatchSpec from skillopt.envs.base import EnvAdapter from skillopt.envs.officeqa.dataloader import OfficeQADataLoader @@ -26,11 +24,11 @@ def __init__( limit: int = 0, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = "offline", + search_mode: str = "auto", max_queries_per_turn: int = 4, - search_api_url: str = os.environ.get("OFFICEQA_SEARCH_API_URL", "http://localhost:8080/search_tool/search"), + search_api_url: str | None = None, search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str = "duckduckgo", + search_provider: str | None = None, search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, @@ -45,9 +43,9 @@ def __init__( self.max_completion_tokens = int(max_completion_tokens) self.search_mode = str(search_mode or "offline") self.max_queries_per_turn = int(max_queries_per_turn) - self.search_api_url = str(search_api_url or "").strip() + self.search_api_url = search_api_url self.search_auth_env = str(search_auth_env or "OFFICEQA_CUSTOM_SEARCH_AUTH").strip() - self.search_provider = str(search_provider or "duckduckgo").strip() + self.search_provider = search_provider self.search_max_num_results = int(search_max_num_results) self.search_timeout_seconds = int(search_timeout_seconds) self.use_local_tools = bool(use_local_tools) diff --git a/skillopt/envs/officeqa/parallel_search.py b/skillopt/envs/officeqa/parallel_search.py new file mode 100644 index 000000000..2b653bd64 --- /dev/null +++ b/skillopt/envs/officeqa/parallel_search.py @@ -0,0 +1,96 @@ +"""Anonymous Parallel Search MCP for the OfficeQA controller's search rounds.""" +from __future__ import annotations + +import asyncio +import json +from datetime import timedelta +from urllib.parse import urlsplit + +import httpx +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client +from mcp.types import Implementation + +from skillopt import __version__ + +MCP_URL = "https://search.parallel.ai/mcp" +USER_AGENT = f"SkillOpt/{__version__} (+https://github.com/microsoft/SkillOpt)" + + +async def _search(queries: list[str], objective: str, session_id: str, timeout: float) -> dict: + # This anonymous connection deliberately does not read API keys or saved MCP + # credentials. Keep the same project identity on every transport request. + async with httpx.AsyncClient( + headers={"User-Agent": USER_AGENT}, + timeout=httpx.Timeout(timeout), + follow_redirects=False, + ) as client: + async with streamable_http_client(MCP_URL, http_client=client) as (read, write, _): + async with ClientSession( + read, write, + read_timeout_seconds=timedelta(seconds=timeout), + client_info=Implementation(name="SkillOpt", version=__version__), + ) as session: + await session.initialize() + tools = await session.list_tools() + if not any(tool.name == "web_search" for tool in tools.tools): + raise RuntimeError("Parallel MCP did not advertise web_search") + result = await session.call_tool("web_search", { + "objective": objective, + "search_queries": queries, + "session_id": session_id, + }) + text = "\n".join(block.text for block in result.content if block.type == "text") + if result.isError: + raise RuntimeError(f"Parallel search error: {text[:1000]}") + payload = result.structuredContent + if payload is None: + try: + payload = json.loads(text) + except (TypeError, ValueError) as exc: + raise RuntimeError("Parallel search returned invalid JSON evidence") from exc + if not isinstance(payload, dict) or not isinstance(payload.get("results"), list): + raise RuntimeError("Parallel search returned invalid results") + return payload + + +def parallel_search( + queries: list[str], *, objective: str, session_id: str, + max_num_results: int = 4, timeout: float = 20, +) -> str: + """Search related queries with a bounded request and render URL evidence. + + The controller supplies one stable session ID per OfficeQA task, including + subsequent rounds. Timeouts and cancellation propagate to the controller. + """ + if not queries or any(not isinstance(query, str) or not query.strip() for query in queries): + raise ValueError("Parallel search requires non-empty queries") + if timeout <= 0 or max_num_results <= 0: + raise ValueError("Parallel search timeout and result limit must be positive") + + async def bounded() -> dict: + return await asyncio.wait_for(_search(queries, objective, session_id, timeout), timeout=timeout) + + try: + payload = asyncio.run(bounded()) + except Exception as exc: + # SDK task groups may wrap a useful transport/tool error. Preserve the + # underlying diagnostic, including explicit anonymous quota responses. + error = exc + while len(getattr(error, "exceptions", ())) == 1: + error = error.exceptions[0] + if error is not exc: + raise error from exc + raise + blocks = [] + for item in payload["results"][:max_num_results]: + if not isinstance(item, dict): + raise RuntimeError("Parallel search returned an invalid result item") + url = item.get("url") + excerpts = item.get("excerpts") + if not isinstance(url, str) or urlsplit(url).scheme not in {"http", "https"} or not urlsplit(url).netloc: + raise RuntimeError("Parallel search returned an invalid result URL") + if not isinstance(excerpts, list) or any(not isinstance(part, str) for part in excerpts): + raise RuntimeError("Parallel search returned invalid excerpts") + blocks.append(f"[{len(blocks) + 1}] {item.get('title') or url}\nURL: {url}\n" + "\n".join(excerpts)[:6000]) + return "\n\n".join(blocks) if blocks else "[no search results]" diff --git a/skillopt/envs/officeqa/rollout.py b/skillopt/envs/officeqa/rollout.py index 7d4c8e045..866efd2b1 100644 --- a/skillopt/envs/officeqa/rollout.py +++ b/skillopt/envs/officeqa/rollout.py @@ -3,10 +3,13 @@ import json import os import re +import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from skillopt.envs.officeqa.evaluator import evaluate +from skillopt.envs.officeqa.parallel_search import parallel_search from skillopt.envs.officeqa.tool_runtime import ( + DEFAULT_CUSTOM_SEARCH_URL, build_oracle_parsed_pages_context, custom_search, resolve_candidate_files, @@ -68,13 +71,31 @@ _DEFAULT_SEARCH_MODE = "offline" _CUSTOM_SEARCH_MODE = "custom_search" _AZURE_SEARCH_MODE = "azure_search" +_PARALLEL_SEARCH_MODE = "parallel" def _normalize_search_mode(search_mode: str | None) -> str: normalized = str(search_mode or _DEFAULT_SEARCH_MODE).strip().lower() + if normalized in {"auto", _PARALLEL_SEARCH_MODE}: + return normalized if normalized in {"custom", _CUSTOM_SEARCH_MODE}: return _CUSTOM_SEARCH_MODE if normalized in {"azure", _AZURE_SEARCH_MODE}: return _AZURE_SEARCH_MODE return _DEFAULT_SEARCH_MODE +def _resolve_search_mode( + search_mode: str | None, *, api_url: str | None, provider: str | None, auth_env: str, + use_local_tools: bool = True, +) -> str: + """Resolve fresh installations without overriding saved modes or services.""" + mode = _normalize_search_mode(search_mode) + if mode != "auto": + return mode + if not use_local_tools: + return _DEFAULT_SEARCH_MODE + if (api_url or os.environ.get("OFFICEQA_SEARCH_API_URL", "")).strip() or (provider or "").strip() or os.environ.get(auth_env, "").strip(): + return _CUSTOM_SEARCH_MODE + # Other harnesses retain their existing local tools and permission controls. + return _PARALLEL_SEARCH_MODE if get_target_backend() in {"openai_chat", "qwen_chat"} else _DEFAULT_SEARCH_MODE + def _build_system( skill_content: str, *, @@ -95,7 +116,7 @@ def _build_system( "answer inside ....\n\n" + skill_section ).rstrip() - if normalized_search_mode == _CUSTOM_SEARCH_MODE: + if normalized_search_mode in {_CUSTOM_SEARCH_MODE, _PARALLEL_SEARCH_MODE}: protocol = ( "You are an expert OfficeQA research assistant. Solve the question using the provided oracle parsed " "OfficeQA page(s) and evidence returned by the controller-managed custom search loop.\n\n" @@ -169,7 +190,7 @@ def _build_user( parts.append("## File Hints\n" + "\n".join(f"- {hint}" for hint in item["source_files"])) if diagnostic_mode and diagnostic_instruction.strip(): parts.append(f"## Training Readout\n{diagnostic_instruction.strip()}") - if normalized_search_mode == _CUSTOM_SEARCH_MODE: + if normalized_search_mode in {_CUSTOM_SEARCH_MODE, _PARALLEL_SEARCH_MODE}: parts.append( _build_round_instruction( turn=turn, @@ -352,16 +373,17 @@ def _run_custom_search_process( search_max_num_results: int, search_timeout_seconds: int, oracle_context: str = "", + search_mode: str = _CUSTOM_SEARCH_MODE, ) -> tuple[str, str, str, str, list[dict], str, dict]: - if not str(search_api_url or "").strip(): + if search_mode == _CUSTOM_SEARCH_MODE and not str(search_api_url or "").strip(): raise ValueError("custom_search mode requires a non-empty search_api_url") - if not os.environ.get(search_auth_env, "").strip(): + if search_mode == _CUSTOM_SEARCH_MODE and not os.environ.get(search_auth_env, "").strip(): raise ValueError(f"custom_search mode requires auth token env var {search_auth_env}") if get_target_backend() not in {"openai_chat", "qwen_chat"}: - raise ValueError("custom_search mode is only supported with target_backend='openai_chat' or 'qwen_chat'") + raise ValueError(f"{search_mode} mode is only supported with target_backend='openai_chat' or 'qwen_chat'") system = _build_system( skill_content, - search_mode=_CUSTOM_SEARCH_MODE, + search_mode=search_mode, max_tool_turns=max_tool_turns, max_queries_per_turn=max_queries_per_turn, ) @@ -369,7 +391,7 @@ def _run_custom_search_process( item, diagnostic_mode=diagnostic_mode, diagnostic_instruction=diagnostic_instruction, - search_mode=_CUSTOM_SEARCH_MODE, + search_mode=search_mode, turn=1, max_tool_turns=max_tool_turns, max_queries_per_turn=max_queries_per_turn, @@ -385,6 +407,7 @@ def _run_custom_search_process( final_answer = "" fail_reason = "" last_response_metadata: dict = {} + session_id = uuid.uuid4().hex for turn in range(1, max_tool_turns + 1): message, _ = chat_target_messages( messages=messages, @@ -411,15 +434,22 @@ def _run_custom_search_process( if not queries: fail_reason = "Model neither produced search queries nor a final answer" break - results_text = _execute_custom_search_round( - queries, - api_url=search_api_url, - auth_env=search_auth_env, - provider=search_provider, - max_num_results=search_max_num_results, - timeout=search_timeout_seconds, - ) - conversation.append({"type": "tool_call", "turn": turn, "cmd": f"custom_search({queries!r})", "obs": results_text}) + if search_mode == _PARALLEL_SEARCH_MODE: + # A failed explicit provider stays an error; never retry via another service. + results_text = parallel_search( + queries, objective=item["question"], session_id=session_id, + max_num_results=search_max_num_results, timeout=search_timeout_seconds, + ) + else: + results_text = _execute_custom_search_round( + queries, + api_url=search_api_url, + auth_env=search_auth_env, + provider=search_provider, + max_num_results=search_max_num_results, + timeout=search_timeout_seconds, + ) + conversation.append({"type": "tool_call", "turn": turn, "cmd": f"{search_mode}({queries!r})", "obs": results_text}) latest_user = ( f"## Search Results Round {turn}\n{results_text}\n\n" + _build_round_instruction( @@ -518,11 +548,11 @@ def process_one( *, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = _DEFAULT_SEARCH_MODE, + search_mode: str = "auto", max_queries_per_turn: int = 4, - search_api_url: str = "", + search_api_url: str | None = None, search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str = "duckduckgo", + search_provider: str | None = None, search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, @@ -533,7 +563,15 @@ def process_one( item_id = str(item["id"]) pred_dir = confined_task_output_dir(out_root, item_id) os.makedirs(pred_dir, exist_ok=True) - normalized_search_mode = _normalize_search_mode(search_mode) + normalized_search_mode = _resolve_search_mode( + search_mode, api_url=search_api_url, provider=search_provider, auth_env=search_auth_env, + use_local_tools=use_local_tools, + ) + search_api_url = str( + os.environ.get("OFFICEQA_SEARCH_API_URL", DEFAULT_CUSTOM_SEARCH_URL) + if search_api_url is None else search_api_url + ).strip() + search_provider = str(search_provider or "duckduckgo").strip() docs_roots: list[str] = [] candidate_files: list[str] = [] oracle_context = "" @@ -550,8 +588,14 @@ def process_one( else "Treat it as primary document evidence for answering the question." ), ) - elif normalized_search_mode == _CUSTOM_SEARCH_MODE: - docs_roots = resolve_docs_roots(data_dirs) + elif normalized_search_mode in {_CUSTOM_SEARCH_MODE, _PARALLEL_SEARCH_MODE}: + try: + docs_roots = resolve_docs_roots(data_dirs) + except FileNotFoundError: + if normalized_search_mode == _CUSTOM_SEARCH_MODE: + raise + # Web evidence works without downloading a local OfficeQA corpus. + docs_roots = [] if item.get("source_files"): candidate_files = resolve_candidate_files(item.get("source_files", []), docs_roots) oracle_context = build_oracle_parsed_pages_context( @@ -583,7 +627,7 @@ def process_one( last_response_metadata: dict = {} allowed_files = [os.path.basename(path) for path in candidate_files] try: - if normalized_search_mode == _CUSTOM_SEARCH_MODE: + if normalized_search_mode in {_CUSTOM_SEARCH_MODE, _PARALLEL_SEARCH_MODE}: system, user, final_response, final_answer, conversation, fail_reason, last_response_metadata = _run_custom_search_process( item, skill_content, @@ -598,6 +642,7 @@ def process_one( search_max_num_results=search_max_num_results, search_timeout_seconds=search_timeout_seconds, oracle_context=oracle_context, + search_mode=normalized_search_mode, ) elif normalized_search_mode == _AZURE_SEARCH_MODE: system, user, final_response, final_answer, conversation, fail_reason, last_response_metadata = _run_azure_search_process( @@ -727,11 +772,11 @@ def run_batch( workers: int = 8, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = _DEFAULT_SEARCH_MODE, + search_mode: str = "auto", max_queries_per_turn: int = 4, - search_api_url: str = "", + search_api_url: str | None = None, search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str = "duckduckgo", + search_provider: str | None = None, search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, diff --git a/tests/test_officeqa_parallel_search.py b/tests/test_officeqa_parallel_search.py new file mode 100644 index 000000000..7c5a73396 --- /dev/null +++ b/tests/test_officeqa_parallel_search.py @@ -0,0 +1,237 @@ +"""OfficeQA caller coverage using the real MCP HTTP transport and scripted model turns.""" +from __future__ import annotations + +import asyncio +import importlib +import json +from types import SimpleNamespace +from unittest import mock + +import httpx +import pytest + +from skillopt.envs.officeqa import rollout +from skillopt.envs.officeqa.adapter import OfficeQAAdapter + +mcp_search = importlib.import_module("skillopt.envs.officeqa.parallel_search") +ITEM = {"id": "treasury", "question": "When was the declaration signed?", "ground_truth": "1776"} +PAYLOAD = {"results": [{"url": "https://example.org/treasury?year=1776", "title": "Treasury history", "excerpts": ["The declaration was signed in 1776."]}]} + + +@pytest.fixture(autouse=True) +def isolated_search_config(monkeypatch): + for name in ("OFFICEQA_CUSTOM_SEARCH_AUTH", "OFFICEQA_SEARCH_API_URL", "PARALLEL_API_KEY"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(rollout, "get_target_backend", lambda: "openai_chat") + + +@pytest.fixture +def transport(monkeypatch): + """Observe actual SDK HTTP requests without substituting its MCP protocol.""" + requests = [] + behavior = {"payload": PAYLOAD, "error": None, "delay": False} + original_client = httpx.AsyncClient + + async def handler(request): + body = json.loads(request.content) if request.content else {} + requests.append((request, body)) + method = body.get("method") + if method == "notifications/initialized": + return httpx.Response(202) + if method == "initialize": + result = {"protocolVersion": "2025-03-26", "capabilities": {"tools": {}}, "serverInfo": {"name": "fixture", "version": "1"}} + elif method == "tools/list": + result = {"tools": [{"name": "web_search", "inputSchema": {"type": "object"}}]} + elif method == "tools/call": + if behavior["delay"]: + await asyncio.Event().wait() + if behavior["error"]: + result = {"content": [{"type": "text", "text": behavior["error"]}], "isError": True} + else: + result = {"content": [{"type": "text", "text": json.dumps(behavior["payload"])}], "structuredContent": behavior["payload"]} + else: + raise AssertionError(f"Unexpected MCP request: {request.method} {body}") + return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": result}) + + def client(**kwargs): + return original_client(transport=httpx.MockTransport(handler), **kwargs) + + monkeypatch.setattr(mcp_search.httpx, "AsyncClient", client) + return requests, behavior + + +def scripted_model(monkeypatch, *, rounds=1): + calls = [] + + def chat(**kwargs): + # Copy now because the caller appends further messages after the call. + messages = [dict(message) for message in kwargs["messages"]] + calls.append(messages) + if len(calls) <= rounds: + text = '["declaration treasury history"]' + else: + assert "https://example.org/treasury?year=1776" in messages[-1]["content"] + assert "signed in 1776" in messages[-1]["content"] + text = "1776" + return SimpleNamespace(content=text), {} + + monkeypatch.setattr(rollout, "chat_target_messages", chat) + return calls + + +@pytest.mark.parametrize("backend", ["openai_chat", "qwen_chat"]) +def test_fresh_adapter_runs_mcp_and_consumes_evidence(tmp_path, monkeypatch, transport, backend): + monkeypatch.setattr(rollout, "get_target_backend", lambda: backend) + calls = scripted_model(monkeypatch, rounds=2) + adapter = OfficeQAAdapter(workers=1, max_tool_turns=3) + rows = adapter.rollout([ITEM], "Read evidence before answering.", str(tmp_path)) + assert rows[0]["predicted_answer"] == "1776" + assert rows[0]["hard"] == 1 + assert len(calls) == 3 + requests, _ = transport + methods = [body.get("method") for _, body in requests] + assert methods.count("initialize") == methods.count("tools/list") == methods.count("tools/call") == 2 + searches = [body["params"] for _, body in requests if body.get("method") == "tools/call"] + assert all(call["name"] == "web_search" for call in searches) + assert searches[0]["arguments"]["session_id"] == searches[1]["arguments"]["session_id"] + assert len(searches[0]["arguments"]["session_id"]) == 32 + assert searches[0]["arguments"]["search_queries"] == ["declaration treasury history"] + assert searches[0]["arguments"]["objective"] == ITEM["question"] + assert all(str(request.url) == "https://search.parallel.ai/mcp" for request, _ in requests) + assert all(request.headers["user-agent"] == mcp_search.USER_AGENT for request, _ in requests) + assert all("authorization" not in request.headers for request, _ in requests) + conversation = json.loads((tmp_path / "predictions" / ITEM["id"] / "conversation.json").read_text()) + assert any(event.get("type") == "tool_call" and "signed in 1776" in event["obs"] for event in conversation) + + +def test_environment_key_is_not_sent_to_anonymous_mcp(monkeypatch, transport): + monkeypatch.setenv("PARALLEL_API_KEY", "must-not-be-sent") + text = mcp_search.parallel_search(["treasury"], objective="history", session_id="a" * 32) + assert "https://example.org/treasury?year=1776" in text + assert all("authorization" not in request.headers for request, _ in transport[0]) + + +def test_advertised_yaml_resolves_through_train_adapter(tmp_path, monkeypatch, transport): + from scripts.train import get_adapter + from skillopt.config import flatten_config, load_config + + cfg = flatten_config(load_config("configs/officeqa/default.yaml")) + cfg.update(workers=1, max_tool_turns=2) + scripted_model(monkeypatch) + adapter = get_adapter(cfg) + assert adapter.search_mode == "auto" + assert adapter.search_provider is None + assert adapter.search_api_url is None + rows = adapter.rollout([ITEM], "", str(tmp_path)) + assert rows[0]["hard"] == 1 + assert any(body.get("method") == "tools/call" for _, body in transport[0]) + + +@pytest.mark.parametrize("mode", ["offline", "custom", "custom_search", "azure", "azure_search", "parallel"]) +def test_explicit_modes_override_ambient_settings(mode, monkeypatch): + monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") + resolved = rollout._resolve_search_mode(mode, api_url="https://configured.example", provider="google", auth_env="OFFICEQA_CUSTOM_SEARCH_AUTH") + assert resolved == {"custom": "custom_search", "azure": "azure_search"}.get(mode, mode) + + +@pytest.mark.parametrize("setting", ["token", "url", "provider", "env_url"]) +def test_auto_preserves_each_configured_service(setting, monkeypatch): + kwargs = {"api_url": None, "provider": None, "auth_env": "OFFICEQA_CUSTOM_SEARCH_AUTH"} + if setting == "token": + monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") + elif setting == "env_url": + monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", "https://configured.example") + else: + kwargs["api_url" if setting == "url" else "provider"] = "https://configured.example" if setting == "url" else "google" + assert rollout._resolve_search_mode("auto", **kwargs) == "custom_search" + + +@pytest.mark.parametrize("backend", ["codex_exec", "claude_code_exec", "cursor_exec", "copilot_exec", "openai_compatible"]) +def test_auto_keeps_other_harness_tool_permissions(backend, monkeypatch): + monkeypatch.setattr(rollout, "get_target_backend", lambda: backend) + assert rollout._resolve_search_mode("auto", api_url=None, provider=None, auth_env="OFFICEQA_CUSTOM_SEARCH_AUTH") == "offline" + + +@pytest.mark.parametrize("mode", ["offline", "auto"]) +def test_explicit_offline_no_tools_makes_no_search(tmp_path, monkeypatch, transport, mode): + docs = tmp_path / "docs" + docs.mkdir() + monkeypatch.setattr(rollout, "chat_target_messages", lambda **kwargs: (SimpleNamespace(content="1776"), {})) + with mock.patch.object(rollout, "parallel_search", side_effect=AssertionError("search disabled")), mock.patch.object(rollout, "run_tool", side_effect=AssertionError("local tools disabled")): + row = rollout.process_one(ITEM, str(tmp_path / "out"), "", search_mode=mode, use_local_tools=False, data_dirs=[str(docs)]) + assert row["predicted_answer"] == "1776" + assert not transport[0] + + +def test_explicit_parallel_ignores_legacy_credentials(tmp_path, monkeypatch, transport): + monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") + monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", "https://configured.example") + scripted_model(monkeypatch) + row = rollout.process_one(ITEM, str(tmp_path), "", search_mode="parallel", search_provider="google", max_tool_turns=2) + assert row["hard"] == 1 + assert transport[0] + assert all("authorization" not in request.headers for request, _ in transport[0]) + + +def test_saved_custom_provider_failure_does_not_use_parallel(tmp_path, monkeypatch, transport): + docs = tmp_path / "docs" + docs.mkdir() + monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") + calls = [] + + def chat(**kwargs): + calls.append(kwargs) + if len(calls) == 1: + return SimpleNamespace(content='["treasury"]'), {} + assert "configured service failed" in kwargs["messages"][-1]["content"] + return SimpleNamespace(content="unknown"), {} + + monkeypatch.setattr(rollout, "chat_target_messages", chat) + with mock.patch.object(rollout, "custom_search", side_effect=RuntimeError("configured service failed")) as custom: + row = rollout.process_one(ITEM, str(tmp_path / "out"), "", search_mode="custom_search", search_api_url="https://configured.example", search_provider="google", data_dirs=[str(docs)], max_tool_turns=2) + assert custom.call_args.kwargs["provider"] == "google" + assert custom.call_args.kwargs["api_url"] == "https://configured.example" + assert not transport[0] + assert row["hard"] == 0 + + +@pytest.mark.parametrize("error", ["anonymous quota exceeded", "upstream unavailable"]) +def test_mcp_errors_fail_task_without_fallback(tmp_path, monkeypatch, transport, error): + transport[1]["error"] = error + calls = scripted_model(monkeypatch) + with mock.patch.object(rollout, "custom_search", side_effect=AssertionError("no fallback")): + row = rollout.process_one(ITEM, str(tmp_path), "", max_tool_turns=2) + assert error in row["fail_reason"] + assert row["hard"] == 0 + assert len(calls) == 1 + + +def test_whole_mcp_request_has_a_deadline(transport): + transport[1]["delay"] = True + with pytest.raises(TimeoutError): + mcp_search.parallel_search(["treasury"], objective="history", session_id="a" * 32, timeout=0.05) + + +def test_cancellation_closes_actual_mcp_session(transport): + transport[1]["delay"] = True + + async def cancel(): + task = asyncio.create_task(mcp_search._search(["treasury"], "history", "a" * 32, 20)) + for _ in range(500): + if any(body.get("method") == "tools/call" for _, body in transport[0]): + break + await asyncio.sleep(0.001) + else: + pytest.fail("MCP caller never reached search") + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(cancel()) + + +@pytest.mark.parametrize("payload", [{"results": [{"url": "file:///etc/passwd", "excerpts": ["bad"]}]}, {"results": "malformed"}, {"results": [{"url": "https://example.org", "excerpts": "not a list"}]}]) +def test_bad_mcp_evidence_is_rejected(transport, payload): + transport[1]["payload"] = payload + with pytest.raises(RuntimeError, match="invalid"): + mcp_search.parallel_search(["treasury"], objective="history", session_id="a" * 32) From 363812322ba670224561a111dc7e19bc9fb6097a Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Sun, 18 Oct 2026 07:15:55 +0000 Subject: [PATCH 2/4] Preserve existing custom search endpoint selection --- docs/reference/config.md | 2 +- skillopt/envs/officeqa/adapter.py | 5 +++++ skillopt/envs/officeqa/rollout.py | 4 +++- tests/test_officeqa_parallel_search.py | 19 ++++++++++++++++++- 4 files changed, 27 insertions(+), 3 deletions(-) diff --git a/docs/reference/config.md b/docs/reference/config.md index 177784a7e..c9ec2842d 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -224,7 +224,7 @@ configuration is still required. | Setting | Behavior | |---|---| -| `env.search_mode: auto` | Use a configured custom service when `search_api_url`, `search_provider`, `OFFICEQA_SEARCH_API_URL`, or the token named by `search_auth_env` is set; otherwise use Parallel for `openai_chat`/`qwen_chat`. Other backends, or `use_local_tools: false`, retain offline mode. | +| `env.search_mode: auto` | Use a configured custom service when `search_api_url`, `search_provider`, a custom `search_auth_env` name, `OFFICEQA_SEARCH_API_URL`, or the token named by `search_auth_env` is set; otherwise use Parallel for `openai_chat`/`qwen_chat`. Other backends, or `use_local_tools: false`, retain offline mode. | | `env.search_mode: parallel` | Explicitly select anonymous Parallel search with `openai_chat`/`qwen_chat`, ignoring custom service settings. | | `env.search_mode: offline` | Keep search local to the configured document corpus, with no Parallel connection. `env.use_local_tools: false` also disables local function tools in this mode. | | `env.search_mode: custom_search` | Keep the existing token-authenticated HTTP service and provider. Set `search_api_url`, `search_auth_env` and `search_provider` for your service. | diff --git a/skillopt/envs/officeqa/adapter.py b/skillopt/envs/officeqa/adapter.py index 37ea47b15..cc858bf4d 100644 --- a/skillopt/envs/officeqa/adapter.py +++ b/skillopt/envs/officeqa/adapter.py @@ -1,5 +1,7 @@ from __future__ import annotations +import os + from skillopt.datasets.base import BatchSpec from skillopt.envs.base import EnvAdapter from skillopt.envs.officeqa.dataloader import OfficeQADataLoader @@ -44,6 +46,9 @@ def __init__( self.search_mode = str(search_mode or "offline") self.max_queries_per_turn = int(max_queries_per_turn) self.search_api_url = search_api_url + if search_api_url is None and self.search_mode.strip().lower() in {"custom", "custom_search"}: + # Preserve the programmatic adapter's existing local-service default. + self.search_api_url = os.environ.get("OFFICEQA_SEARCH_API_URL", "http://localhost:8080/search_tool/search") self.search_auth_env = str(search_auth_env or "OFFICEQA_CUSTOM_SEARCH_AUTH").strip() self.search_provider = search_provider self.search_max_num_results = int(search_max_num_results) diff --git a/skillopt/envs/officeqa/rollout.py b/skillopt/envs/officeqa/rollout.py index 866efd2b1..f5e92f2c4 100644 --- a/skillopt/envs/officeqa/rollout.py +++ b/skillopt/envs/officeqa/rollout.py @@ -91,7 +91,9 @@ def _resolve_search_mode( return mode if not use_local_tools: return _DEFAULT_SEARCH_MODE - if (api_url or os.environ.get("OFFICEQA_SEARCH_API_URL", "")).strip() or (provider or "").strip() or os.environ.get(auth_env, "").strip(): + if ((api_url or os.environ.get("OFFICEQA_SEARCH_API_URL", "")).strip() + or (provider or "").strip() or auth_env != "OFFICEQA_CUSTOM_SEARCH_AUTH" + or os.environ.get(auth_env, "").strip()): return _CUSTOM_SEARCH_MODE # Other harnesses retain their existing local tools and permission controls. return _PARALLEL_SEARCH_MODE if get_target_backend() in {"openai_chat", "qwen_chat"} else _DEFAULT_SEARCH_MODE diff --git a/tests/test_officeqa_parallel_search.py b/tests/test_officeqa_parallel_search.py index 7c5a73396..7b062d459 100644 --- a/tests/test_officeqa_parallel_search.py +++ b/tests/test_officeqa_parallel_search.py @@ -134,18 +134,35 @@ def test_explicit_modes_override_ambient_settings(mode, monkeypatch): assert resolved == {"custom": "custom_search", "azure": "azure_search"}.get(mode, mode) -@pytest.mark.parametrize("setting", ["token", "url", "provider", "env_url"]) +@pytest.mark.parametrize("setting", ["token", "url", "provider", "env_url", "auth_env"]) def test_auto_preserves_each_configured_service(setting, monkeypatch): kwargs = {"api_url": None, "provider": None, "auth_env": "OFFICEQA_CUSTOM_SEARCH_AUTH"} if setting == "token": monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") elif setting == "env_url": monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", "https://configured.example") + elif setting == "auth_env": + kwargs["auth_env"] = "MY_SEARCH_TOKEN" else: kwargs["api_url" if setting == "url" else "provider"] = "https://configured.example" if setting == "url" else "google" assert rollout._resolve_search_mode("auto", **kwargs) == "custom_search" +@pytest.mark.parametrize("url", [None, "https://configured.example/search"]) +def test_legacy_adapter_preserves_omitted_custom_url(tmp_path, monkeypatch, transport, url): + docs = tmp_path / "docs" + docs.mkdir() + monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") + if url: + monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", url) + scripted_model(monkeypatch) + with mock.patch.object(rollout, "custom_search", return_value="URL: https://example.org/treasury?year=1776\nThe declaration was signed in 1776.") as custom: + rows = OfficeQAAdapter(search_mode="custom_search", workers=1, max_tool_turns=2, data_dirs=[str(docs)]).rollout([ITEM], "", str(tmp_path / "out")) + assert rows[0]["hard"] == 1 + assert custom.call_args.kwargs["api_url"] == (url or "http://localhost:8080/search_tool/search") + assert not transport[0] + + @pytest.mark.parametrize("backend", ["codex_exec", "claude_code_exec", "cursor_exec", "copilot_exec", "openai_compatible"]) def test_auto_keeps_other_harness_tool_permissions(backend, monkeypatch): monkeypatch.setattr(rollout, "get_target_backend", lambda: backend) From e980f57c67c37bcc1caf316c8696ba39c0181c24 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 18 Oct 2026 17:52:29 +0000 Subject: [PATCH 3/4] Make Parallel search an explicit opt-in with an optional MCP dependency Addresses review feedback on the default selection and the core dependency. Restore the offline default: configs/officeqa/default.yaml and OfficeQAAdapter are byte-identical to main again, and rollout no longer resolves an `auto` mode. `auto` (like any unrecognized value) normalizes to `offline`, so a fresh configuration behaves exactly as it does on main. Parallel is reached only through an explicit `env.search_mode=parallel`. Move `mcp` out of core dependencies into a `parallel` extra and import the Parallel client lazily, so OfficeQA imports and offline installs no longer require it. Selecting the mode without the extra raises an error naming `pip install 'skillopt[parallel]'`. CI installs `.[dev,parallel]` so the provider stays covered. Document the data boundary in docs/reference/config.md and the changelog: selecting this mode sends the generated queries and the full task question (as the MCP `objective`) to Parallel's hosted endpoint. Describe the keyless tier as an anonymous, lower-rate free tier rather than unlimited search, and keep the note that this evidence is not directly comparable to the paper's Google-backed setting. Tests cover that a fresh `auto` config does not select Parallel, that explicit selection does, that the OfficeQA env imports with `mcp` absent, and that selecting Parallel without the extra reports the install command. MCP transport tests skip when the extra is not installed. Co-authored-by: George Pickett --- .github/workflows/ci.yml | 4 +- CHANGELOG.md | 13 +- configs/officeqa/default.yaml | 7 +- docs/reference/config.md | 60 ++++--- pyproject.toml | 3 +- skillopt/envs/officeqa/adapter.py | 13 +- skillopt/envs/officeqa/parallel_search.py | 8 +- skillopt/envs/officeqa/rollout.py | 59 +++---- tests/test_officeqa_parallel_search.py | 187 +++++++++++++--------- 9 files changed, 199 insertions(+), 155 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8cabfd7ed..b7824c235 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,7 +19,9 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install - run: pip install -e ".[dev]" + # `parallel` is an opt-in extra, but the OfficeQA Parallel search tests + # skip without it, so CI installs it to keep that provider covered. + run: pip install -e ".[dev,parallel]" - name: Pytest run: python -m pytest -q env: diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ee77088f..47d0e99a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,10 +7,15 @@ All notable changes to SkillOpt are documented here. This project adheres to ## [Unreleased] ### Added -- **OfficeQA Parallel Search MCP**: fresh `auto` search configurations use - anonymous Fast-mode web search with `openai_chat`/`qwen_chat` when no custom - service is configured. Saved offline, custom-service and Azure search modes - remain explicit choices; other harnesses retain their local tool permissions. +- **OfficeQA Parallel Search MCP** as an opt-in `env.search_mode=parallel` + provider for `openai_chat`/`qwen_chat`. The default stays `offline` and no + config selects Parallel automatically. Selecting it sends the generated + search queries and the full task question (as the MCP `objective`) to + Parallel's hosted endpoint, over an anonymous, lower-rate free tier, so + its scores are not directly comparable to the paper's Google-backed search + setting. The MCP client ships in a new `parallel` extra + (`pip install 'skillopt[parallel]'`) and is imported lazily, so OfficeQA + runs and offline installs do not require `mcp`. - **SkillOpt-Sleep paired A/B evalkit** (`python -m skillopt_sleep.evalkit`): McNemar plus percentile-bootstrap CIs on a fixed task manifest, with task-cluster multi-seed inference and seeded null calibration for exact-test diff --git a/configs/officeqa/default.yaml b/configs/officeqa/default.yaml index e4c2be458..7b72f1a4b 100644 --- a/configs/officeqa/default.yaml +++ b/configs/officeqa/default.yaml @@ -24,12 +24,11 @@ env: workers: 4 max_tool_turns: 24 max_completion_tokens: 16384 - # Fresh chat runs use keyless Parallel; saved offline/custom/Azure modes remain explicit. - search_mode: auto + search_mode: offline max_queries_per_turn: 4 - search_api_url: null + search_api_url: http://apisix.westus2.cloudapp.azure.com/search_tool/search search_auth_env: OFFICEQA_CUSTOM_SEARCH_AUTH - search_provider: null + search_provider: duckduckgo search_max_num_results: 4 search_timeout_seconds: 20 limit: 0 diff --git a/docs/reference/config.md b/docs/reference/config.md index c9ec2842d..0645c0a67 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -212,35 +212,47 @@ Benchmark-specific `env` keys are passed through to the adapter. ### OfficeQA web search -New OfficeQA configs use `env.search_mode: auto`. With `openai_chat` or -`qwen_chat`, an unconfigured search service uses -[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) -at `https://search.parallel.ai/mcp`. Search runs in Fast mode without an -account or API key. The controller sends the model's `` to -MCP and supplies the returned URLs and excerpts to the next model round. -Parallel search does not require a local OfficeQA document corpus; parsed -source-page context is included when that corpus is available. Model backend -configuration is still required. +`env.search_mode` selects how an OfficeQA rollout gathers evidence. The +default is unchanged: `offline` searches only the configured local document +corpus and makes no network search call. | Setting | Behavior | |---|---| -| `env.search_mode: auto` | Use a configured custom service when `search_api_url`, `search_provider`, a custom `search_auth_env` name, `OFFICEQA_SEARCH_API_URL`, or the token named by `search_auth_env` is set; otherwise use Parallel for `openai_chat`/`qwen_chat`. Other backends, or `use_local_tools: false`, retain offline mode. | -| `env.search_mode: parallel` | Explicitly select anonymous Parallel search with `openai_chat`/`qwen_chat`, ignoring custom service settings. | -| `env.search_mode: offline` | Keep search local to the configured document corpus, with no Parallel connection. `env.use_local_tools: false` also disables local function tools in this mode. | -| `env.search_mode: custom_search` | Keep the existing token-authenticated HTTP service and provider. Set `search_api_url`, `search_auth_env` and `search_provider` for your service. | -| `env.search_mode: azure_search` | Keep the existing model-provided Azure search path. | +| `env.search_mode: offline` (default) | Keep search local to the configured document corpus. `env.use_local_tools: false` also disables local function tools in this mode. | +| `env.search_mode: custom_search` | Use the token-authenticated HTTP service. Set `search_api_url`, `search_auth_env` and `search_provider` for your service. | +| `env.search_mode: azure_search` | Use the model-provided Azure search path. | +| `env.search_mode: parallel` | Opt in to Parallel's hosted Search MCP. Requires `openai_chat` or `qwen_chat` and the `parallel` extra. | | `env.search_timeout_seconds` | Whole MCP discovery/search deadline in seconds (default `20`). | | `env.search_max_num_results` | Maximum number of Parallel evidence results per round (default `4`). | -| `env.max_queries_per_turn` / `env.max_tool_turns` | Bound model-requested queries and model rounds as before. | - -Existing saved `offline`, `custom_search`, and `azure_search` settings remain -authoritative. To keep a new experiment local, set `env.search_mode=offline` -with `--cfg-options`. To switch a saved config to Parallel, set -`env.search_mode=parallel`. Explicit providers never fall back to Parallel on -failure. Anonymous requests do not use `PARALLEL_API_KEY` or saved MCP -credentials; service errors, including free-tier limits, fail the rollout -with a search diagnostic. This provider changes the evidence source, so its -scores are not a reproduction of the paper's Google-backed search setting. +| `env.max_queries_per_turn` / `env.max_tool_turns` | Bound model-requested queries and model rounds. | + +#### Parallel Search MCP (opt-in) + +[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) +is never selected automatically. It is used only when you set +`env.search_mode=parallel` explicitly, for example with +`--cfg-options env.search_mode=parallel`. It also needs the optional MCP +client, installed with `pip install 'skillopt[parallel]'`; selecting the mode +without that extra raises an error naming the install command. Every other +mode, and every OfficeQA import, works without `mcp` installed. + +**Data boundary.** When this mode is selected, each search round sends data +to Parallel's hosted endpoint at `https://search.parallel.ai/mcp`: the +model's generated ``, and the **full text of the task +question** as the MCP `objective` argument. Do not select this mode for +tasks whose question text cannot leave your environment. Requests are +anonymous — they carry no `PARALLEL_API_KEY` and no saved MCP credentials — +so they run on the anonymous, lower-rate free tier described in Parallel's +documentation. That tier is rate-limited, not unmetered: expect throttling on +sustained runs, and use a keyed provider for large sweeps. Rate-limit and +other service errors fail +the rollout with a search diagnostic; an explicit provider never silently +falls back to another one. + +Parallel search does not require a local OfficeQA document corpus; parsed +source-page context is included when that corpus is available. This mode +changes the evidence source, so its scores are not directly comparable to, +and are not a reproduction of, the paper's Google-backed search setting. ## Credential Environment Variables diff --git a/pyproject.toml b/pyproject.toml index c3cd9b450..3808b69e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,6 @@ dependencies = [ "azure-identity>=1.15.0", "azure-core>=1.30.0", "httpx>=0.27.0", - "mcp>=1.27.0,<2", ] [project.optional-dependencies] @@ -43,6 +42,8 @@ claude = ["claude-agent-sdk>=0.1.0", "json_repair>=0.61.0"] qwen = ["vllm>=0.4.0", "json_repair>=0.61.0"] # SearchQA data materialization searchqa = ["datasets>=2.18.0"] +# OfficeQA `search_mode: parallel` web search over Parallel's hosted Search MCP +parallel = ["mcp>=1.27.0,<2"] # Documentation site docs = ["mkdocs-material>=9.5.0", "mkdocstrings[python]>=0.24.0"] # WebUI dashboard diff --git a/skillopt/envs/officeqa/adapter.py b/skillopt/envs/officeqa/adapter.py index cc858bf4d..63419d48f 100644 --- a/skillopt/envs/officeqa/adapter.py +++ b/skillopt/envs/officeqa/adapter.py @@ -26,11 +26,11 @@ def __init__( limit: int = 0, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = "auto", + search_mode: str = "offline", max_queries_per_turn: int = 4, - search_api_url: str | None = None, + search_api_url: str = os.environ.get("OFFICEQA_SEARCH_API_URL", "http://localhost:8080/search_tool/search"), search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str | None = None, + search_provider: str = "duckduckgo", search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, @@ -45,12 +45,9 @@ def __init__( self.max_completion_tokens = int(max_completion_tokens) self.search_mode = str(search_mode or "offline") self.max_queries_per_turn = int(max_queries_per_turn) - self.search_api_url = search_api_url - if search_api_url is None and self.search_mode.strip().lower() in {"custom", "custom_search"}: - # Preserve the programmatic adapter's existing local-service default. - self.search_api_url = os.environ.get("OFFICEQA_SEARCH_API_URL", "http://localhost:8080/search_tool/search") + self.search_api_url = str(search_api_url or "").strip() self.search_auth_env = str(search_auth_env or "OFFICEQA_CUSTOM_SEARCH_AUTH").strip() - self.search_provider = search_provider + self.search_provider = str(search_provider or "duckduckgo").strip() self.search_max_num_results = int(search_max_num_results) self.search_timeout_seconds = int(search_timeout_seconds) self.use_local_tools = bool(use_local_tools) diff --git a/skillopt/envs/officeqa/parallel_search.py b/skillopt/envs/officeqa/parallel_search.py index 2b653bd64..add326b0e 100644 --- a/skillopt/envs/officeqa/parallel_search.py +++ b/skillopt/envs/officeqa/parallel_search.py @@ -1,4 +1,10 @@ -"""Anonymous Parallel Search MCP for the OfficeQA controller's search rounds.""" +"""Parallel Search MCP client for the OfficeQA controller's search rounds. + +Reached only through the explicit `search_mode=parallel` selection. Every call +sends the model's generated queries and the full task question (as the MCP +`objective`) to Parallel's hosted endpoint, anonymously, on the documented +lower-rate free tier. +""" from __future__ import annotations import asyncio diff --git a/skillopt/envs/officeqa/rollout.py b/skillopt/envs/officeqa/rollout.py index f5e92f2c4..e45629cd5 100644 --- a/skillopt/envs/officeqa/rollout.py +++ b/skillopt/envs/officeqa/rollout.py @@ -7,9 +7,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from skillopt.envs.officeqa.evaluator import evaluate -from skillopt.envs.officeqa.parallel_search import parallel_search from skillopt.envs.officeqa.tool_runtime import ( - DEFAULT_CUSTOM_SEARCH_URL, build_oracle_parsed_pages_context, custom_search, resolve_candidate_files, @@ -74,30 +72,27 @@ _PARALLEL_SEARCH_MODE = "parallel" def _normalize_search_mode(search_mode: str | None) -> str: normalized = str(search_mode or _DEFAULT_SEARCH_MODE).strip().lower() - if normalized in {"auto", _PARALLEL_SEARCH_MODE}: - return normalized if normalized in {"custom", _CUSTOM_SEARCH_MODE}: return _CUSTOM_SEARCH_MODE if normalized in {"azure", _AZURE_SEARCH_MODE}: return _AZURE_SEARCH_MODE + if normalized == _PARALLEL_SEARCH_MODE: + return _PARALLEL_SEARCH_MODE return _DEFAULT_SEARCH_MODE -def _resolve_search_mode( - search_mode: str | None, *, api_url: str | None, provider: str | None, auth_env: str, - use_local_tools: bool = True, -) -> str: - """Resolve fresh installations without overriding saved modes or services.""" - mode = _normalize_search_mode(search_mode) - if mode != "auto": - return mode - if not use_local_tools: - return _DEFAULT_SEARCH_MODE - if ((api_url or os.environ.get("OFFICEQA_SEARCH_API_URL", "")).strip() - or (provider or "").strip() or auth_env != "OFFICEQA_CUSTOM_SEARCH_AUTH" - or os.environ.get(auth_env, "").strip()): - return _CUSTOM_SEARCH_MODE - # Other harnesses retain their existing local tools and permission controls. - return _PARALLEL_SEARCH_MODE if get_target_backend() in {"openai_chat", "qwen_chat"} else _DEFAULT_SEARCH_MODE +def _load_parallel_search(): + """Import the Parallel client only when `parallel` mode is actually selected. + The `mcp` package is an optional extra, so importing it at module scope + would make it a hard requirement of every OfficeQA rollout. + """ + try: + from skillopt.envs.officeqa.parallel_search import parallel_search + except ImportError as exc: + raise RuntimeError( + "search_mode='parallel' requires the optional MCP client. " + "Install it with: pip install 'skillopt[parallel]'" + ) from exc + return parallel_search def _build_system( skill_content: str, *, @@ -438,7 +433,7 @@ def _run_custom_search_process( break if search_mode == _PARALLEL_SEARCH_MODE: # A failed explicit provider stays an error; never retry via another service. - results_text = parallel_search( + results_text = _load_parallel_search()( queries, objective=item["question"], session_id=session_id, max_num_results=search_max_num_results, timeout=search_timeout_seconds, ) @@ -550,11 +545,11 @@ def process_one( *, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = "auto", + search_mode: str = _DEFAULT_SEARCH_MODE, max_queries_per_turn: int = 4, - search_api_url: str | None = None, + search_api_url: str = "", search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str | None = None, + search_provider: str = "duckduckgo", search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, @@ -565,15 +560,7 @@ def process_one( item_id = str(item["id"]) pred_dir = confined_task_output_dir(out_root, item_id) os.makedirs(pred_dir, exist_ok=True) - normalized_search_mode = _resolve_search_mode( - search_mode, api_url=search_api_url, provider=search_provider, auth_env=search_auth_env, - use_local_tools=use_local_tools, - ) - search_api_url = str( - os.environ.get("OFFICEQA_SEARCH_API_URL", DEFAULT_CUSTOM_SEARCH_URL) - if search_api_url is None else search_api_url - ).strip() - search_provider = str(search_provider or "duckduckgo").strip() + normalized_search_mode = _normalize_search_mode(search_mode) docs_roots: list[str] = [] candidate_files: list[str] = [] oracle_context = "" @@ -774,11 +761,11 @@ def run_batch( workers: int = 8, max_tool_turns: int = 12, max_completion_tokens: int = 16384, - search_mode: str = "auto", + search_mode: str = _DEFAULT_SEARCH_MODE, max_queries_per_turn: int = 4, - search_api_url: str | None = None, + search_api_url: str = "", search_auth_env: str = "OFFICEQA_CUSTOM_SEARCH_AUTH", - search_provider: str | None = None, + search_provider: str = "duckduckgo", search_max_num_results: int = 4, search_timeout_seconds: int = 20, use_local_tools: bool = True, diff --git a/tests/test_officeqa_parallel_search.py b/tests/test_officeqa_parallel_search.py index 7b062d459..22ad76f97 100644 --- a/tests/test_officeqa_parallel_search.py +++ b/tests/test_officeqa_parallel_search.py @@ -1,19 +1,24 @@ -"""OfficeQA caller coverage using the real MCP HTTP transport and scripted model turns.""" +"""OfficeQA search-mode selection, optional-dependency handling and MCP caller coverage.""" from __future__ import annotations import asyncio +import builtins import importlib +import importlib.util import json +import sys from types import SimpleNamespace from unittest import mock -import httpx import pytest from skillopt.envs.officeqa import rollout from skillopt.envs.officeqa.adapter import OfficeQAAdapter -mcp_search = importlib.import_module("skillopt.envs.officeqa.parallel_search") +HAS_MCP = importlib.util.find_spec("mcp") is not None +requires_mcp = pytest.mark.skipif(not HAS_MCP, reason="requires the optional 'parallel' extra") +mcp_search = importlib.import_module("skillopt.envs.officeqa.parallel_search") if HAS_MCP else None + ITEM = {"id": "treasury", "question": "When was the declaration signed?", "ground_truth": "1776"} PAYLOAD = {"results": [{"url": "https://example.org/treasury?year=1776", "title": "Treasury history", "excerpts": ["The declaration was signed in 1776."]}]} @@ -28,6 +33,8 @@ def isolated_search_config(monkeypatch): @pytest.fixture def transport(monkeypatch): """Observe actual SDK HTTP requests without substituting its MCP protocol.""" + import httpx + requests = [] behavior = {"payload": PAYLOAD, "error": None, "delay": False} original_client = httpx.AsyncClient @@ -79,11 +86,100 @@ def chat(**kwargs): return calls +def block_mcp_import(monkeypatch): + """Make `import mcp` fail the way an install without the extra would.""" + real_import = builtins.__import__ + + def guarded(name, *args, **kwargs): + if name == "mcp" or name.startswith("mcp."): + raise ImportError("No module named 'mcp'") + return real_import(name, *args, **kwargs) + + for name in [key for key in sys.modules if key == "mcp" or key.startswith("mcp.")]: + monkeypatch.delitem(sys.modules, name) + monkeypatch.delitem(sys.modules, "skillopt.envs.officeqa.parallel_search", raising=False) + monkeypatch.setattr(builtins, "__import__", guarded) + + +# ── Default selection: Parallel is never automatic ─────────────────────────── + +def test_shipped_config_keeps_the_offline_default(): + from scripts.train import get_adapter + from skillopt.config import flatten_config, load_config + + cfg = flatten_config(load_config("configs/officeqa/default.yaml")) + assert cfg["search_mode"] == "offline" + adapter = get_adapter(cfg) + assert adapter.search_mode == "offline" + assert adapter.search_provider == "duckduckgo" + assert rollout._normalize_search_mode(adapter.search_mode) == rollout._DEFAULT_SEARCH_MODE + + +@pytest.mark.parametrize("mode", ["auto", "", None, "fresh", "web", "parallel_search"]) +def test_auto_and_unrecognized_modes_never_select_parallel(mode): + assert rollout._normalize_search_mode(mode) == rollout._DEFAULT_SEARCH_MODE + + +@pytest.mark.parametrize("mode", ["parallel", "PARALLEL", " Parallel "]) +def test_parallel_requires_explicit_selection(mode): + assert rollout._normalize_search_mode(mode) == rollout._PARALLEL_SEARCH_MODE + + +@pytest.mark.parametrize( + ("mode", "expected"), + [("offline", "offline"), ("custom", "custom_search"), ("custom_search", "custom_search"), ("azure", "azure_search"), ("azure_search", "azure_search")], +) +def test_existing_modes_resolve_unchanged(mode, expected): + assert rollout._normalize_search_mode(mode) == expected + + +@pytest.mark.parametrize("mode", ["auto", "offline"]) +def test_fresh_auto_config_runs_offline_without_search(tmp_path, monkeypatch, mode): + docs = tmp_path / "docs" + docs.mkdir() + monkeypatch.setattr(rollout, "chat_target_messages", lambda **kwargs: (SimpleNamespace(content="1776"), {})) + with ( + mock.patch.object(rollout, "_load_parallel_search", side_effect=AssertionError("parallel not selected")), + mock.patch.object(rollout, "custom_search", side_effect=AssertionError("custom search not selected")), + mock.patch.object(rollout, "run_tool", side_effect=AssertionError("local tools disabled")), + ): + row = rollout.process_one(ITEM, str(tmp_path / "out"), "", search_mode=mode, use_local_tools=False, data_dirs=[str(docs)]) + assert row["predicted_answer"] == "1776" + + +# ── Optional `mcp` dependency ──────────────────────────────────────────────── + +def test_officeqa_env_imports_without_mcp(monkeypatch): + block_mcp_import(monkeypatch) + for name in [key for key in sys.modules if key.startswith("skillopt.envs.officeqa")]: + monkeypatch.delitem(sys.modules, name) + reloaded = importlib.import_module("skillopt.envs.officeqa.adapter") + assert reloaded.OfficeQAAdapter().search_mode == "offline" + assert importlib.import_module("skillopt.envs.officeqa.rollout")._normalize_search_mode("offline") == "offline" + + +def test_parallel_without_the_extra_names_the_install_command(monkeypatch): + block_mcp_import(monkeypatch) + with pytest.raises(RuntimeError, match=r"skillopt\[parallel\]"): + rollout._load_parallel_search() + + +def test_parallel_rollout_without_the_extra_fails_with_the_install_command(tmp_path, monkeypatch): + block_mcp_import(monkeypatch) + scripted_model(monkeypatch) + row = rollout.process_one(ITEM, str(tmp_path), "", search_mode="parallel", max_tool_turns=2) + assert "skillopt[parallel]" in row["fail_reason"] + assert row["hard"] == 0 + + +# ── Parallel MCP caller ────────────────────────────────────────────────────── + +@requires_mcp @pytest.mark.parametrize("backend", ["openai_chat", "qwen_chat"]) -def test_fresh_adapter_runs_mcp_and_consumes_evidence(tmp_path, monkeypatch, transport, backend): +def test_explicit_parallel_runs_mcp_and_consumes_evidence(tmp_path, monkeypatch, transport, backend): monkeypatch.setattr(rollout, "get_target_backend", lambda: backend) calls = scripted_model(monkeypatch, rounds=2) - adapter = OfficeQAAdapter(workers=1, max_tool_turns=3) + adapter = OfficeQAAdapter(search_mode="parallel", workers=1, max_tool_turns=3) rows = adapter.rollout([ITEM], "Read evidence before answering.", str(tmp_path)) assert rows[0]["predicted_answer"] == "1776" assert rows[0]["hard"] == 1 @@ -96,6 +192,7 @@ def test_fresh_adapter_runs_mcp_and_consumes_evidence(tmp_path, monkeypatch, tra assert searches[0]["arguments"]["session_id"] == searches[1]["arguments"]["session_id"] assert len(searches[0]["arguments"]["session_id"]) == 32 assert searches[0]["arguments"]["search_queries"] == ["declaration treasury history"] + # Documented data boundary: the whole task question leaves as the objective. assert searches[0]["arguments"]["objective"] == ITEM["question"] assert all(str(request.url) == "https://search.parallel.ai/mcp" for request, _ in requests) assert all(request.headers["user-agent"] == mcp_search.USER_AGENT for request, _ in requests) @@ -104,6 +201,7 @@ def test_fresh_adapter_runs_mcp_and_consumes_evidence(tmp_path, monkeypatch, tra assert any(event.get("type") == "tool_call" and "signed in 1776" in event["obs"] for event in conversation) +@requires_mcp def test_environment_key_is_not_sent_to_anonymous_mcp(monkeypatch, transport): monkeypatch.setenv("PARALLEL_API_KEY", "must-not-be-sent") text = mcp_search.parallel_search(["treasury"], objective="history", session_id="a" * 32) @@ -111,75 +209,7 @@ def test_environment_key_is_not_sent_to_anonymous_mcp(monkeypatch, transport): assert all("authorization" not in request.headers for request, _ in transport[0]) -def test_advertised_yaml_resolves_through_train_adapter(tmp_path, monkeypatch, transport): - from scripts.train import get_adapter - from skillopt.config import flatten_config, load_config - - cfg = flatten_config(load_config("configs/officeqa/default.yaml")) - cfg.update(workers=1, max_tool_turns=2) - scripted_model(monkeypatch) - adapter = get_adapter(cfg) - assert adapter.search_mode == "auto" - assert adapter.search_provider is None - assert adapter.search_api_url is None - rows = adapter.rollout([ITEM], "", str(tmp_path)) - assert rows[0]["hard"] == 1 - assert any(body.get("method") == "tools/call" for _, body in transport[0]) - - -@pytest.mark.parametrize("mode", ["offline", "custom", "custom_search", "azure", "azure_search", "parallel"]) -def test_explicit_modes_override_ambient_settings(mode, monkeypatch): - monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") - resolved = rollout._resolve_search_mode(mode, api_url="https://configured.example", provider="google", auth_env="OFFICEQA_CUSTOM_SEARCH_AUTH") - assert resolved == {"custom": "custom_search", "azure": "azure_search"}.get(mode, mode) - - -@pytest.mark.parametrize("setting", ["token", "url", "provider", "env_url", "auth_env"]) -def test_auto_preserves_each_configured_service(setting, monkeypatch): - kwargs = {"api_url": None, "provider": None, "auth_env": "OFFICEQA_CUSTOM_SEARCH_AUTH"} - if setting == "token": - monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") - elif setting == "env_url": - monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", "https://configured.example") - elif setting == "auth_env": - kwargs["auth_env"] = "MY_SEARCH_TOKEN" - else: - kwargs["api_url" if setting == "url" else "provider"] = "https://configured.example" if setting == "url" else "google" - assert rollout._resolve_search_mode("auto", **kwargs) == "custom_search" - - -@pytest.mark.parametrize("url", [None, "https://configured.example/search"]) -def test_legacy_adapter_preserves_omitted_custom_url(tmp_path, monkeypatch, transport, url): - docs = tmp_path / "docs" - docs.mkdir() - monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") - if url: - monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", url) - scripted_model(monkeypatch) - with mock.patch.object(rollout, "custom_search", return_value="URL: https://example.org/treasury?year=1776\nThe declaration was signed in 1776.") as custom: - rows = OfficeQAAdapter(search_mode="custom_search", workers=1, max_tool_turns=2, data_dirs=[str(docs)]).rollout([ITEM], "", str(tmp_path / "out")) - assert rows[0]["hard"] == 1 - assert custom.call_args.kwargs["api_url"] == (url or "http://localhost:8080/search_tool/search") - assert not transport[0] - - -@pytest.mark.parametrize("backend", ["codex_exec", "claude_code_exec", "cursor_exec", "copilot_exec", "openai_compatible"]) -def test_auto_keeps_other_harness_tool_permissions(backend, monkeypatch): - monkeypatch.setattr(rollout, "get_target_backend", lambda: backend) - assert rollout._resolve_search_mode("auto", api_url=None, provider=None, auth_env="OFFICEQA_CUSTOM_SEARCH_AUTH") == "offline" - - -@pytest.mark.parametrize("mode", ["offline", "auto"]) -def test_explicit_offline_no_tools_makes_no_search(tmp_path, monkeypatch, transport, mode): - docs = tmp_path / "docs" - docs.mkdir() - monkeypatch.setattr(rollout, "chat_target_messages", lambda **kwargs: (SimpleNamespace(content="1776"), {})) - with mock.patch.object(rollout, "parallel_search", side_effect=AssertionError("search disabled")), mock.patch.object(rollout, "run_tool", side_effect=AssertionError("local tools disabled")): - row = rollout.process_one(ITEM, str(tmp_path / "out"), "", search_mode=mode, use_local_tools=False, data_dirs=[str(docs)]) - assert row["predicted_answer"] == "1776" - assert not transport[0] - - +@requires_mcp def test_explicit_parallel_ignores_legacy_credentials(tmp_path, monkeypatch, transport): monkeypatch.setenv("OFFICEQA_CUSTOM_SEARCH_AUTH", "configured-token") monkeypatch.setenv("OFFICEQA_SEARCH_API_URL", "https://configured.example") @@ -190,6 +220,7 @@ def test_explicit_parallel_ignores_legacy_credentials(tmp_path, monkeypatch, tra assert all("authorization" not in request.headers for request, _ in transport[0]) +@requires_mcp def test_saved_custom_provider_failure_does_not_use_parallel(tmp_path, monkeypatch, transport): docs = tmp_path / "docs" docs.mkdir() @@ -212,23 +243,26 @@ def chat(**kwargs): assert row["hard"] == 0 -@pytest.mark.parametrize("error", ["anonymous quota exceeded", "upstream unavailable"]) +@requires_mcp +@pytest.mark.parametrize("error", ["anonymous rate limit exceeded", "upstream unavailable"]) def test_mcp_errors_fail_task_without_fallback(tmp_path, monkeypatch, transport, error): transport[1]["error"] = error calls = scripted_model(monkeypatch) with mock.patch.object(rollout, "custom_search", side_effect=AssertionError("no fallback")): - row = rollout.process_one(ITEM, str(tmp_path), "", max_tool_turns=2) + row = rollout.process_one(ITEM, str(tmp_path), "", search_mode="parallel", max_tool_turns=2) assert error in row["fail_reason"] assert row["hard"] == 0 assert len(calls) == 1 +@requires_mcp def test_whole_mcp_request_has_a_deadline(transport): transport[1]["delay"] = True with pytest.raises(TimeoutError): mcp_search.parallel_search(["treasury"], objective="history", session_id="a" * 32, timeout=0.05) +@requires_mcp def test_cancellation_closes_actual_mcp_session(transport): transport[1]["delay"] = True @@ -247,6 +281,7 @@ async def cancel(): asyncio.run(cancel()) +@requires_mcp @pytest.mark.parametrize("payload", [{"results": [{"url": "file:///etc/passwd", "excerpts": ["bad"]}]}, {"results": "malformed"}, {"results": [{"url": "https://example.org", "excerpts": "not a list"}]}]) def test_bad_mcp_evidence_is_rejected(transport, payload): transport[1]["payload"] = payload From babd828b8fb7a1d6468e01a18170a3894f661cdf Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:48:02 +0000 Subject: [PATCH 4/4] Raise the built-in TimeoutError when the MCP deadline fires The whole-request deadline fired on every supported version, but callers saw two different exception types. `asyncio.wait_for` raises `asyncio.exceptions.TimeoutError`, which only became an alias of the built-in `TimeoutError` in 3.11, so the strict deadline assertion failed the CI Python 3.10 job while 3.11 and 3.12 passed. Normalize in the provider instead of loosening the assertion: inside the existing task-group unwrapping, re-raise an exceeded deadline as the built-in `TimeoutError` with the original timeout error as its cause. This covers both a direct `wait_for` timeout and one surfaced from an SDK task group, and leaves the deadline itself and the cancellation-cleanup path untouched. --- skillopt/envs/officeqa/parallel_search.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/skillopt/envs/officeqa/parallel_search.py b/skillopt/envs/officeqa/parallel_search.py index add326b0e..0a9d4d42e 100644 --- a/skillopt/envs/officeqa/parallel_search.py +++ b/skillopt/envs/officeqa/parallel_search.py @@ -67,7 +67,8 @@ def parallel_search( """Search related queries with a bounded request and render URL evidence. The controller supplies one stable session ID per OfficeQA task, including - subsequent rounds. Timeouts and cancellation propagate to the controller. + subsequent rounds. Timeouts and cancellation propagate to the controller; + an exceeded deadline always raises the built-in `TimeoutError`. """ if not queries or any(not isinstance(query, str) or not query.strip() for query in queries): raise ValueError("Parallel search requires non-empty queries") @@ -85,6 +86,10 @@ async def bounded() -> dict: error = exc while len(getattr(error, "exceptions", ())) == 1: error = error.exceptions[0] + if isinstance(error, asyncio.TimeoutError): + # Before 3.11 `asyncio.TimeoutError` is its own class rather than the + # built-in one, so normalize the deadline for callers on every version. + raise TimeoutError(f"Parallel search exceeded its {timeout}s deadline") from exc if error is not exc: raise error from exc raise