diff --git a/.changeset/olive-moons-search.md b/.changeset/olive-moons-search.md new file mode 100644 index 00000000..07a4f319 --- /dev/null +++ b/.changeset/olive-moons-search.md @@ -0,0 +1,5 @@ +--- +"apollo": minor +--- + +global_chat: opt-in web search and fetch for the planner diff --git a/services/global_chat/PAYLOAD_SPEC.md b/services/global_chat/PAYLOAD_SPEC.md index 6fa64018..2d080329 100644 --- a/services/global_chat/PAYLOAD_SPEC.md +++ b/services/global_chat/PAYLOAD_SPEC.md @@ -40,7 +40,8 @@ This document defines the input and output payload structure for the Global Agen ], "options": { // Runtime options (optional) - "stream": false + "stream": false, + "web_search": false }, "api_key": "string (REQUIRED in production, optional in development)" @@ -82,6 +83,7 @@ This document defines the input and output payload structure for the Global Agen - **`options`** (object, optional): Runtime options. - **`stream`** (boolean): Enable streaming response (default: false). + - **`web_search`** (boolean): Let the planner search and fetch pages on the live web for this request (default: `false`). Takes effect **only on planner-routed requests** — the direct `workflow_agent` / `job_code_agent` routes ignore it, and `meta.web_search_requested` records when it was set on a request that never reached the planner. Reachable domains are limited to a server-side allowlist. Requires the caller's own Anthropic key to have web search enabled in their Anthropic Console; searches bill to that key, and clients on a zero-data-retention contract cannot use it. If the key does not have it enabled, the turn still answers — without web results — and sets `meta.web_search_downgraded`. - **`api_key`** (string, **required in production**, optional in development): API key for the Anthropic API. In production environments this field is required and requests without it will be rejected. In development, the server falls back to the `ANTHROPIC_API_KEY` environment variable if this field is omitted. @@ -131,7 +133,20 @@ This document defines the input and output payload structure for the Global Agen { "tool": "call_workflow_agent", "input": { "message": "..." } } ], "subagent_calls": [], // Raw sub-agent result dicts (for debugging) - "total_tool_calls": 2 + "total_tool_calls": 2, + + // Only when the planner gave up mid-turn: + "truncated": true, + "stop_reason": "pause_turn", + + // Only when options.web_search was set: + "web_search_requested": true, + + // Only when the planner has web tools on: + "web_searches": 2, + "web_fetches": 2, + "web_domains": ["hl7.org", "docs.openfn.org"], + "web_search_downgraded": false } } ``` @@ -168,6 +183,11 @@ Each tool beat streams as: `thinking` spinner → `changes` (if the workflow was - **`tool_calls`** (array): List of `{tool, input}` objects for each tool the planner invoked (planner path only). - **`subagent_calls`** (array): Raw sub-agent result dicts including `_call_metadata`. On the planner path these are the full results, useful for debugging. On the router's direct job-code path it carries a single entry with just `_call_metadata` and `diff`, so a client can tell on either route whether a code edit actually landed (`diff.patches_applied`). - **`total_tool_calls`** (number): Total number of tool calls made by the planner (planner path only). + - **`truncated`** (boolean): `true` when the planner spent its `max_pause_continuations` budget while the API still had more of the turn to send `response` is the head of a reply the server split and not a finished answer. Accompanied by **`stop_reason`** (`"pause_turn"`). + - **`web_search_requested`** (boolean): Present and `true` only when the request set `options.web_search`. + - **`web_searches`** / **`web_fetches`** (number): Server-side web search and web fetch calls the planner made this turn. + - **`web_domains`** (array): Hostnames the planner fetched from this turn, deduplicated. + - **`web_search_downgraded`** (boolean): `true` when the web tools were dropped mid-turn because the caller's Anthropic key rejected them, and the turn was answered without web results. --- diff --git a/services/global_chat/config.yaml b/services/global_chat/config.yaml index 112b13e9..ca394f25 100644 --- a/services/global_chat/config.yaml +++ b/services/global_chat/config.yaml @@ -11,3 +11,11 @@ planner: model: "claude-opus" max_tokens: 24576 max_tool_calls: 20 + max_pause_continuations: 5 + web_search: # Server-side web search/fetch for planner. + max_uses: 5 + max_content_tokens: 10000 + # Content from these domains steers the agent that edits workflows. + allowed_domains: + - hl7.org + - docs.openfn.org diff --git a/services/global_chat/global_chat.py b/services/global_chat/global_chat.py index 1e9c11cd..443fde75 100644 --- a/services/global_chat/global_chat.py +++ b/services/global_chat/global_chat.py @@ -61,6 +61,10 @@ def get_stream(self) -> bool: """Extract stream flag from options.""" return (self.options or {}).get("stream", False) + def get_web_search(self) -> bool: + """Extract web_search flag from options.""" + return (self.options or {}).get("web_search", False) + @observe(name="global_chat", capture_input=False) def main(data_dict: dict) -> dict: @@ -108,6 +112,7 @@ def main(data_dict: dict) -> dict: attachments=data.attachments or [], user=user_info, metrics_opt_in=data.metrics_opt_in, + web_search=data.get_web_search(), ) if tracking: diff --git a/services/global_chat/planner.py b/services/global_chat/planner.py index 42ba790a..aa1fd95e 100644 --- a/services/global_chat/planner.py +++ b/services/global_chat/planner.py @@ -3,12 +3,13 @@ """ import os +from urllib.parse import urlparse from typing import List, Dict, Optional from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass import httpx import anthropic -from anthropic import Anthropic +from anthropic import Anthropic, BadRequestError import sentry_sdk import sys @@ -23,10 +24,11 @@ STATUS_REVIEWING_WORKFLOW, STATUS_NEW_WORKFLOW, STATUS_PLANNING, + STATUS_SEARCHING_WEB, ) from global_chat.config_loader import ConfigLoader from models import resolve_model -from global_chat.tools.tool_definitions import TOOL_DEFINITIONS +from global_chat.tools.tool_definitions import TOOL_DEFINITIONS, build_web_tools from yaml_utils import stitch_job_code, redact_job_bodies, find_job_in_yaml, get_step_name_from_page, inspect_job_code, job_keys_in_yaml from tools.search_documentation.search_documentation import search_documentation_tool from global_chat.subagent_caller import call_workflow_agent, call_job_agent, format_subagent_result_for_llm @@ -221,7 +223,12 @@ class PlannerAgent: Planner agent that coordinates subagents and tools for complex multi-step tasks. """ - def __init__(self, config_loader: ConfigLoader, api_key: Optional[str] = None): + def __init__( + self, + config_loader: ConfigLoader, + api_key: Optional[str] = None, + web_search: bool = False, + ): self.config_loader = config_loader self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY") @@ -229,12 +236,20 @@ def __init__(self, config_loader: ConfigLoader, api_key: Optional[str] = None): raise ApolloError(500, "ANTHROPIC_API_KEY not found") self.client = Anthropic(api_key=self.api_key) - self.tools = TOOL_DEFINITIONS + + self.web_tools = build_web_tools(config_loader.config) if web_search else [] + self.web_search_enabled = bool(self.web_tools) + self.web_search_downgraded = False + self.tools = TOOL_DEFINITIONS + self.web_tools + + if web_search and not self.web_tools: + logger.info("web_search requested but no allowed_domains configured, web tools are disabled") planner_config = config_loader.config.get("planner", {}) self.model = resolve_model(planner_config.get("model", "claude-opus")) self.max_tokens = planner_config.get("max_tokens", 24576) self.max_tool_calls = planner_config.get("max_tool_calls", 20) + self.max_pause_continuations = planner_config.get("max_pause_continuations", 5) self.current_yaml: Optional[str] = None self.subagent_results = [] @@ -303,6 +318,9 @@ def run( tool_call_count = 0 tool_calls_meta = [] + paused_text = "" + pause_count = 0 + web_usage = {"web_searches": 0, "web_fetches": 0, "web_domains": []} total_usage = { "input_tokens": 0, "output_tokens": 0, @@ -317,13 +335,40 @@ def run( while not final_round: final_round = tool_call_count >= self.max_tool_calls try: - response = self._call_api( - system_prompt, - messages, - stream, - stream_manager, - tool_choice={"type": "none"} if final_round else None, - ) + try: + response = self._call_api( + system_prompt, + messages, + stream, + stream_manager, + tool_choice={"type": "none"} if final_round else None, + ) + except BadRequestError as web_error: + # We retry without the web tools to see if they were the cause + # as the 400 body carries no machine-readable reason. + if not self.web_tools: + raise + logger.warning(f"BadRequestError with the web tools active, retrying without them: {web_error}") + self.web_tools = [] + self.tools = TOOL_DEFINITIONS + self.web_search_downgraded = True + system_prompt = self._build_system_prompt() + try: + response = self._call_api( + system_prompt, + messages, + stream, + stream_manager, + tool_choice={"type": "none"} if final_round else None, + ) + except BadRequestError: + # The web tools were not the cause. Surface the + # original error. + raise web_error from None + self._send_settled( + stream_manager, + "Web search is unavailable for this account — answering without it", + ) for field in [ "input_tokens", @@ -333,6 +378,13 @@ def run( ]: total_usage[field] += getattr(response.usage, field, 0) + round_web = self._count_server_tool_uses(response) + web_usage["web_searches"] += round_web["web_searches"] + web_usage["web_fetches"] += round_web["web_fetches"] + for host in round_web["web_domains"]: + if host not in web_usage["web_domains"]: + web_usage["web_domains"].append(host) + logger.info(f"Claude API call {tool_call_count + 1}: stop_reason={response.stop_reason}") # Text from every round is part of the answer the user saw @@ -359,28 +411,26 @@ def run( tool_use_blocks, stream_manager, total_usage, tool_calls_meta ) - content_blocks = [] - for block in response.content: - if block.type == "thinking": - content_blocks.append({ - "type": "thinking", - "thinking": block.thinking, - "signature": block.signature, - }) - elif block.type == "text": - content_blocks.append({"type": "text", "text": block.text}) - elif block.type == "tool_use": - content_blocks.append( - {"type": "tool_use", "id": block.id, "name": block.name, "input": block.input} - ) - tool_call_count += len(tool_use_blocks) if tool_call_count >= self.max_tool_calls: tool_results.append({"type": "text", "text": _FINAL_ROUND_NOTICE}) - messages.append({"role": "assistant", "content": content_blocks}) + # Append the response's own blocks rather than a whitelist of known types + messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": tool_results}) + paused_text = "" + + elif response.stop_reason == "pause_turn": + messages.append({"role": "assistant", "content": response.content}) + paused_text += round_text + round_text = "" + pause_count += 1 + if pause_count >= self.max_pause_continuations: + logger.warning(f"Pause budget spent after {pause_count} continuations") + break + continue + else: logger.warning(f"Unexpected stop_reason: {response.stop_reason}") break @@ -440,7 +490,9 @@ def run( # response and history keep only the last round's text (the actual # answer), matching the direct routes and what was saved before # narration was streamed. The narration survives in response_segments. - final_text = round_text + # This does not apply to paused_text, a pause_turn round is the same answer + # the server split, so its head belongs to the final text. + final_text = paused_text + round_text if not final_text: stop_reason = getattr(response, "stop_reason", None) @@ -478,19 +530,29 @@ def run( return_history.append({"role": "user", "content": content}) return_history.append({"role": "assistant", "content": final_text}) + meta = { + "agents": agents_used, + "planner_iterations": tool_call_count, + "tool_calls": tool_calls_meta, + "subagent_calls": self.subagent_results, + "total_tool_calls": tool_call_count, + } + + if response.stop_reason == "pause_turn": + meta["truncated"] = True + meta["stop_reason"] = "pause_turn" + + if self.web_search_enabled: + meta.update(web_usage) + meta["web_search_downgraded"] = self.web_search_downgraded + return PlannerResult( response=final_text, response_segments=response_segments, attachments=attachments, history=return_history, usage=total_usage, - meta={ - "agents": agents_used, - "planner_iterations": tool_call_count, - "tool_calls": tool_calls_meta, - "subagent_calls": self.subagent_results, - "total_tool_calls": tool_call_count, - }, + meta=meta, ) def _build_user_content(self, content: str, page: Optional[str]) -> str: @@ -541,6 +603,7 @@ def _call_api(self, system_prompt, messages, stream, stream_manager, tool_choice choice = {"tool_choice": tool_choice} if tool_choice else {} if stream: + last_settled = None with self.client.beta.messages.stream( model=self.model, max_tokens=self.max_tokens, @@ -555,6 +618,16 @@ def _call_api(self, system_prompt, messages, stream, stream_manager, tool_choice for event in stream_obj: if event.type == "content_block_delta" and event.delta.type == "text_delta": stream_manager.send_text(event.delta.text) + elif event.type == "content_block_start": + if event.content_block.type == "server_tool_use": + self._send_spinner(stream_manager, STATUS_SEARCHING_WEB) + elif event.type == "content_block_stop": + block = getattr(event, "content_block", None) + if getattr(block, "type", None) in self.WEB_RESULT_BLOCK_TYPES: + message = self._web_result_message(block) + if message and message != last_settled: + self._send_settled(stream_manager, message) + last_settled = message return stream_obj.get_final_message() else: response = self.client.beta.messages.create( @@ -972,8 +1045,67 @@ def _extract_text(self, response): """Extract text from response content, concatenated as it was streamed.""" return "".join(block.text for block in response.content if block.type == "text") + @staticmethod + def _count_server_tool_uses(response) -> dict: + """Count web search/fetch uses in one response and note fetched hosts.""" + searches = 0 + fetches = 0 + hosts: list[str] = [] + + for block in response.content: + if getattr(block, "type", None) != "server_tool_use": + continue + if block.name == "web_search": + searches += 1 + elif block.name == "web_fetch": + fetches += 1 + host = urlparse((block.input or {}).get("url", "")).hostname + if host and host not in hosts: + hosts.append(host) + + return {"web_searches": searches, "web_fetches": fetches, "web_domains": hosts} + + WEB_RESULT_BLOCK_TYPES = ("web_search_tool_result", "web_fetch_tool_result") + + WEB_RESULT_BLOCKED_CODES = ("url_not_allowed", "url_not_in_prior_context") + + @staticmethod + def _web_result_error_code(block: object) -> str | None: + """The error_code of a finished web result block, or None when it succeeded.""" + content = getattr(block, "content", None) + if isinstance(content, list): + return None + if isinstance(content, dict): + if str(content.get("type") or "").endswith("_error"): + return content.get("error_code") + return None + if str(getattr(content, "type", "") or "").endswith("_error"): + return getattr(content, "error_code", None) + return None + + @staticmethod + def _web_result_message(block: object) -> str | None: + """The settled line for one finished web result block, or None to say nothing.""" + if getattr(block, "content", None) is None: + return None + error_code = PlannerAgent._web_result_error_code(block) + if error_code is None: + if getattr(block, "type", None) == "web_fetch_tool_result": + return "Read a page from the web" + return "Searched the web" + if error_code in PlannerAgent.WEB_RESULT_BLOCKED_CODES: + return "Skipped a page outside the allowed sources" + return "A web lookup did not return anything" + def _build_system_prompt(self) -> list: """Build system prompt for planner with cache control.""" prompt_text = self.config_loader.get_prompt("planner_system_prompt") - return [{"type": "text", "text": prompt_text, "cache_control": {"type": "ephemeral"}}] + blocks = [{"type": "text", "text": prompt_text, "cache_control": {"type": "ephemeral"}}] + + web_prompt = self.config_loader.get_prompt("planner_web_tools_prompt") + if self.web_tools and web_prompt: + domains = ", ".join(self.web_tools[0].get("allowed_domains") or []) + blocks.append({"type": "text", "text": web_prompt.replace("{domains}", domains)}) + + return blocks diff --git a/services/global_chat/prompts.yaml b/services/global_chat/prompts.yaml index 26d9f845..ec578188 100644 --- a/services/global_chat/prompts.yaml +++ b/services/global_chat/prompts.yaml @@ -187,3 +187,16 @@ prompts: If it says it lacks something you hold, such as an attachment you did not name, call again with it. Only add context if the subagent's response is genuinely unclear or incomplete. + + planner_web_tools_prompt: | + ## Web Search and Fetch + + You also have `web_search` and `web_fetch`, which read pages on the live web. + + - `search_documentation` stays your first choice for OpenFn concepts, adaptors, and platform features. It is faster and more accurate for those than the web. + - Use `web_search` / `web_fetch` for external target-system API references that OpenFn's own docs do not cover: request/response shapes, field names, endpoint paths. + - If `search_documentation` comes back thin or off-target, follow up on the web rather than answering from memory. If OpenFn's own docsite is in the reachable list below, it is there for exactly that case: the search index is a periodic snapshot of chunks, so the live page can be newer or more complete. Never go to the web for an OpenFn topic first. + - When an answer came from a fetched page, name the source in prose. + - You can only reach these domains: {domains}. Nothing else is fetchable. + - For FHIR, use the published R4 spec under `https://hl7.org/fhir/R4/` (e.g. `https://hl7.org/fhir/R4/patient.html`). If you used a different FHIR version, say which one. + - Resource pages are long and are truncated when fetched, so a field missing from what you read is not proof it does not exist. Say what you confirmed and what you could not. diff --git a/services/global_chat/router.py b/services/global_chat/router.py index c2431862..987da279 100644 --- a/services/global_chat/router.py +++ b/services/global_chat/router.py @@ -85,6 +85,7 @@ def route_and_execute( attachments: Optional[List[Dict]] = None, user: Optional[Dict] = None, metrics_opt_in: Optional[bool] = None, + web_search: bool = False, ) -> RouterResult: """ Route request to appropriate handler and execute. @@ -96,6 +97,8 @@ def route_and_execute( history: Conversation history stream: Streaming flag attachments: Optional input attachments (e.g. logs, dataclips) + web_search: Whether the caller opted into the planner's web + search/fetch tools for this request Returns: RouterResult with response, attachments, history, usage, meta @@ -112,6 +115,7 @@ def route_and_execute( self._input_attachments = attachments or [] self._user = user self._metrics_opt_in = metrics_opt_in + self._web_search = web_search # One stream manager shared by whichever agents serve this request, so # a handed-over request continues the same stream instead of starting # a second message lifecycle. @@ -145,6 +149,9 @@ def route_and_execute( else: result = self._route_to_planner(content, workflow_yaml, page, history, stream, decision.confidence) + if web_search: + result.meta["web_search_requested"] = True + return result @observe(name="routing_decision") @@ -481,7 +488,7 @@ def _route_to_planner( clean_history = [{"role": t["role"], "content": t["content"]} for t in history] - planner = PlannerAgent(self.config_loader, self.api_key) + planner = PlannerAgent(self.config_loader, self.api_key, web_search=self._web_search) planner_result = planner.run( content=content, workflow_yaml=workflow_yaml, diff --git a/services/global_chat/tests/unit/test_payload.py b/services/global_chat/tests/unit/test_payload.py new file mode 100644 index 00000000..a46fcaf1 --- /dev/null +++ b/services/global_chat/tests/unit/test_payload.py @@ -0,0 +1,17 @@ +"""Unit tests for the global_chat request payload.""" + +from global_chat.global_chat import Payload + + +def test_web_search_defaults_to_off_when_options_are_absent() -> None: + assert Payload.from_dict({"content": "hi"}).get_web_search() is False + + +def test_web_search_defaults_to_off_when_options_omit_the_key() -> None: + assert Payload.from_dict({"content": "hi", "options": {"stream": True}}).get_web_search() is False + + +def test_web_search_is_read_from_options() -> None: + payload = Payload.from_dict({"content": "hi", "options": {"web_search": True}}) + + assert payload.get_web_search() is True diff --git a/services/global_chat/tests/unit/test_planner.py b/services/global_chat/tests/unit/test_planner.py index 81fa0567..70fb8766 100644 --- a/services/global_chat/tests/unit/test_planner.py +++ b/services/global_chat/tests/unit/test_planner.py @@ -1,19 +1,26 @@ """Unit tests for PlannerAgent tool execution and user-content building.""" +from pathlib import Path from unittest.mock import patch import anthropic import httpx import pytest +import yaml +from anthropic import BadRequestError from util import ApolloError +import global_chat.planner as planner_module from global_chat.planner import ( _FINAL_ROUND_NOTICE, _api_error_message, PlannerAgent, + PlannerResult, ) +from global_chat.tools.tool_definitions import TOOL_DEFINITIONS, build_web_tools +from streaming_util import STATUS_SEARCHING_WEB WORKFLOW_YAML = """\ name: wf @@ -38,6 +45,9 @@ def make_planner() -> PlannerAgent: planner.api_key = "test-key" planner._user = None planner._metrics_opt_in = None + planner.web_tools = [] + planner.web_search_enabled = False + planner.web_search_downgraded = False return planner @@ -61,10 +71,12 @@ def __init__(self, name: str, tool_input: dict, block_id: str = "tu_1"): class StubStreamManager: def __init__(self) -> None: + self.thinking: list = [] self.statuses: list[dict] = [] + self.text: list[str] = [] - def send_thinking(self, *_args: object, **_kwargs: object) -> None: - pass + def send_thinking(self, status: object = None, *_args: object, **_kwargs: object) -> None: + self.thinking.append(status) def send_changes(self, *_args: object, **_kwargs: object) -> None: pass @@ -79,6 +91,57 @@ def send_status( {"content": content, "steps": steps, "summary": summary}, ) + def send_text(self, chunk: str) -> None: + self.text.append(chunk) + + +class FakeTextBlock: + type = "text" + + def __init__(self, text: str) -> None: + self.text = text + + +class FakeUsage: + input_tokens = 0 + output_tokens = 0 + cache_creation_input_tokens = 0 + cache_read_input_tokens = 0 + + +class FakeServerToolUse: + type = "server_tool_use" + + def __init__(self, name: str, tool_input: dict, block_id: str = "srvtu_1") -> None: + self.name = name + self.input = tool_input + self.id = block_id + + +class FakeResponse: + def __init__(self, stop_reason: str, content: list) -> None: + self.stop_reason = stop_reason + self.content = content + self.usage = FakeUsage() + + +def make_run_planner(max_tool_calls: int = 10, max_pause_continuations: int = 5) -> PlannerAgent: + """A planner wired for run(), with no config, client, or tools.""" + planner = make_planner() + planner.model = "claude-test" + planner.max_tokens = 1024 + planner.max_tool_calls = max_tool_calls + planner.max_pause_continuations = max_pause_continuations + planner.tools = [] + return planner + + +def run_with(planner: PlannerAgent, responses: list, content: str = "q") -> PlannerResult: + """Drive planner.run() over a scripted list of API responses.""" + with patch.object(PlannerAgent, "_build_system_prompt", return_value=[]), \ + patch.object(PlannerAgent, "_call_api", side_effect=list(responses)): + return planner.run(content, None, None, [], stream=False) + def test_inspect_job_code_accepts_multiple_keys() -> None: planner = make_planner() @@ -260,6 +323,102 @@ def fake_call_job_agent(_tool_input: dict, workflow_yaml: str, *_args: object, * assert "newCode();" in planner.current_yaml +def test_pause_turn_keeps_the_text_from_before_the_pause() -> None: + planner = make_run_planner() + responses = [ + FakeResponse("pause_turn", [FakeTextBlock("Half an answer. ")]), + FakeResponse("end_turn", [FakeTextBlock("The rest.")]), + ] + + result = run_with(planner, responses) + + assert result.response == "Half an answer. The rest." + assert result.history[-1]["content"] == "Half an answer. The rest." + assert [s["content"] for s in result.response_segments] == ["Half an answer. ", "The rest."] + + +def test_paused_text_survives_the_pause_budget_exit_without_duplicating() -> None: + """Exiting the loop while still paused should keep the head exactly once.""" + planner = make_run_planner(max_pause_continuations=2) + responses = [ + FakeResponse("pause_turn", [FakeTextBlock("A")]), + FakeResponse("pause_turn", [FakeTextBlock("B")]), + ] + + result = run_with(planner, responses) + + assert result.response == "AB" + # Not a finished answer: the server had more of the turn to send. + assert result.meta["truncated"] is True + assert result.meta["stop_reason"] == "pause_turn" + + +def test_pauses_do_not_spend_the_subagent_budget() -> None: + """A paused round made no tool call, so max_tool_calls must be untouched.""" + planner = make_run_planner(max_tool_calls=2) + responses = [ + FakeResponse("pause_turn", [FakeTextBlock("Still working. ")]), + FakeResponse("pause_turn", [FakeTextBlock("Nearly. ")]), + FakeResponse("pause_turn", [FakeTextBlock("Almost. ")]), + FakeResponse("end_turn", [FakeTextBlock("Done.")]), + ] + + result = run_with(planner, responses) + + assert result.response == "Still working. Nearly. Almost. Done." + assert result.meta["planner_iterations"] == 0 + assert "truncated" not in result.meta + + +def test_a_completed_turn_is_not_flagged_as_truncated() -> None: + planner = make_run_planner() + responses = [FakeResponse("end_turn", [FakeTextBlock("Plain answer.")])] + + result = run_with(planner, responses) + + assert "truncated" not in result.meta + assert "stop_reason" not in result.meta + + +def test_a_real_tool_round_resets_the_paused_text_buffer() -> None: + """Narration from before a tool call is not part of the final answer.""" + planner = make_run_planner() + responses = [ + FakeResponse("pause_turn", [FakeTextBlock("Stale narration. ")]), + FakeResponse("tool_use", [FakeToolUse("search_documentation", {"query": "dhis2"})]), + FakeResponse("end_turn", [FakeTextBlock("The real answer.")]), + ] + + with patch("global_chat.planner.search_documentation_tool", return_value="docs"): + result = run_with(planner, responses) + + assert result.response == "The real answer." + + +def test_a_mixed_round_keeps_server_tool_blocks_in_history() -> None: + """A round with both a web search and a local tool should not drop the search blocks.""" + planner = make_run_planner() + search_block = FakeServerToolUse("web_search", {"query": "dhis2 tracker api"}) + responses = [ + FakeResponse("tool_use", [search_block, FakeToolUse("search_documentation", {"query": "x"})]), + FakeResponse("end_turn", [FakeTextBlock("Done.")]), + ] + seen = [] + + def record_and_reply(_system: object, messages: list, _stream: object, _manager: object, **_kwargs: object) -> FakeResponse: + seen.append(list(messages)) + return responses.pop(0) + + with patch.object(PlannerAgent, "_build_system_prompt", return_value=[]), \ + patch.object(PlannerAgent, "_call_api", side_effect=record_and_reply), \ + patch("global_chat.planner.search_documentation_tool", return_value="docs"): + planner.run("q", None, None, [], stream=False) + + assistant_turn = seen[1][-2] + assert assistant_turn["role"] == "assistant" + assert search_block in assistant_turn["content"] + + def test_user_content_names_the_step_being_viewed() -> None: planner = make_planner() @@ -887,3 +1046,549 @@ def test_parallel_batch_overshooting_the_budget_still_ends_on_an_answer() -> Non assert tool_choices == [None, {"type": "none"}] assert notices == [False, True] assert response == "Here is what changed." + + +class StubConfigLoader: + """Minimal ConfigLoader stand-in.""" + + def __init__(self, config: dict) -> None: + self.config = config + + +class StubPromptLoader: + """ConfigLoader stand-in for _build_system_prompt, which only calls get_prompt.""" + + def __init__(self, **prompts: str) -> None: + self._prompts = prompts + + def get_prompt(self, key: str) -> str: + return self._prompts.get(key, "") + + +WEB_CONFIG = { + "planner": { + "model": "claude-opus", + "web_search": { + "max_uses": 5, + "max_content_tokens": 10000, + "allowed_domains": ["docs.dhis2.org"], + }, + }, +} + + +def build_planner(config: dict, *, web_search: bool) -> PlannerAgent: + """Construct a real PlannerAgent with the Anthropic client stubbed out.""" + with patch("global_chat.planner.Anthropic"): + return PlannerAgent(StubConfigLoader(config), api_key="test-key", web_search=web_search) + + +def test_web_tools_are_off_unless_the_request_asks_for_them() -> None: + planner = build_planner(WEB_CONFIG, web_search=False) + + assert planner.web_tools == [] + assert planner.web_search_enabled is False + assert planner.tools == TOOL_DEFINITIONS + + +def test_web_tools_are_appended_after_the_existing_tools() -> None: + planner = build_planner(WEB_CONFIG, web_search=True) + + assert planner.tools[: len(TOOL_DEFINITIONS)] == TOOL_DEFINITIONS + assert [t["name"] for t in planner.tools[len(TOOL_DEFINITIONS):]] == ["web_search", "web_fetch"] + assert planner.web_search_enabled is True + + +def test_the_module_level_tool_list_is_never_mutated() -> None: + before = list(TOOL_DEFINITIONS) + + build_planner(WEB_CONFIG, web_search=True) + + assert before == TOOL_DEFINITIONS + + +def test_an_empty_allowlist_keeps_the_tools_off_even_when_requested() -> None: + planner = build_planner({"planner": {"web_search": {"allowed_domains": []}}}, web_search=True) + + assert planner.web_tools == [] + assert planner.web_search_enabled is False + + +def test_system_prompt_is_a_single_cached_block_without_web_tools() -> None: + planner = make_run_planner() + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + + blocks = planner._build_system_prompt() + + assert len(blocks) == 1 + assert blocks[0]["cache_control"] == {"type": "ephemeral"} + + +def test_system_prompt_appends_an_uncached_web_block_with_the_allowlist() -> None: + planner = make_run_planner() + planner.config_loader = StubPromptLoader( + planner_system_prompt="BASE PROMPT", + planner_web_tools_prompt="WEB ADDENDUM {domains}", + ) + planner.web_tools = build_web_tools(WEB_CONFIG) + + blocks = planner._build_system_prompt() + + # Two blocks, and only the first is cached: the addendum must stay after + # the breakpoint, outside the cache key. + assert [("cache_control" in block) for block in blocks] == [True, False] + assert blocks[0]["text"] == "BASE PROMPT" + assert blocks[0]["cache_control"] == {"type": "ephemeral"} + assert blocks[1]["text"] == "WEB ADDENDUM docs.dhis2.org" + + +def test_the_web_tools_prompt_key_exists_in_prompts_yaml() -> None: + path = Path(planner_module.__file__).parent / "prompts.yaml" + prompts = yaml.safe_load(path.read_text(encoding="utf-8"))["prompts"] + + assert "{domains}" in prompts["planner_web_tools_prompt"] + + +def test_no_web_block_is_appended_when_the_prompt_is_missing() -> None: + """An empty text block would be rejected by the API, so drop it.""" + planner = make_run_planner() + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + planner.web_tools = build_web_tools(WEB_CONFIG) + + blocks = planner._build_system_prompt() + + assert len(blocks) == 1 + + +class FakeEvent: + def __init__(self, event_type: str, **fields: object) -> None: + self.type = event_type + for key, value in fields.items(): + setattr(self, key, value) + + +class FakeBlockRef: + def __init__(self, block_type: str) -> None: + self.type = block_type + + +class FakeStream: + def __init__(self, events: list, final: FakeResponse) -> None: + self._events = events + self._final = final + + def __enter__(self) -> "FakeStream": + return self + + def __exit__(self, *_exc: object) -> bool: + return False + + def __iter__(self): + return iter(self._events) + + def get_final_message(self) -> FakeResponse: + return self._final + + +class FakeMessages: + def __init__(self, stream: FakeStream) -> None: + self._stream = stream + + def stream(self, **_kwargs: object) -> FakeStream: + return self._stream + + +class FakeClient: + def __init__(self, stream: FakeStream) -> None: + self.messages = FakeMessages(stream) + self.beta = self + + +def block_start(block_type: str) -> FakeEvent: + return FakeEvent("content_block_start", content_block=FakeBlockRef(block_type)) + + +def block_stop(block_type: str, content: object) -> FakeEvent: + block = FakeBlockRef(block_type) + block.content = content + return FakeEvent("content_block_stop", content_block=block) + + +def text_delta(text: str) -> FakeEvent: + return FakeEvent("content_block_delta", delta=FakeEvent("text_delta", text=text)) + + +SEARCH_OK = [{"type": "web_search_result", "url": "https://hl7.org/fhir/R4/patient.html"}] +FETCH_OK = {"type": "web_fetch_result", "url": "https://hl7.org/fhir/R4/patient.html"} +FETCH_BLOCKED = {"type": "web_fetch_tool_result_error", "error_code": "url_not_allowed"} +SEARCH_EXHAUSTED = {"type": "web_search_tool_result_error", "error_code": "max_uses_exceeded"} + + +def test_repeated_successful_lookups_settle_as_one_line() -> None: + """Two successful searches in one round should still read as one line.""" + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Answer.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_search_tool_result", SEARCH_OK), + block_start("server_tool_use"), + block_stop("web_search_tool_result", SEARCH_OK), + text_delta("Answer."), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.thinking == [STATUS_SEARCHING_WEB, STATUS_SEARCHING_WEB] + assert manager.statuses == [ + {"content": "Searched the web", "steps": None, "summary": None}, + ] + assert planner._segments == [{"type": "status", "content": "Searched the web"}] + + +def test_a_blocked_fetch_is_not_reported_as_a_successful_lookup() -> None: + """url_not_allowed comes back 200 with an error block, it must not say we read it.""" + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Answering from memory.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_fetch_tool_result", FETCH_BLOCKED), + text_delta("Answering from memory."), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.statuses == [ + {"content": "Skipped a page outside the allowed sources", "steps": None, "summary": None}, + ] + assert planner._segments == [ + {"type": "status", "content": "Skipped a page outside the allowed sources"} + ] + + +def test_a_failed_search_says_so_rather_than_claiming_a_result() -> None: + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("No luck.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_search_tool_result", SEARCH_EXHAUSTED), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.statuses == [ + {"content": "A web lookup did not return anything", "steps": None, "summary": None}, + ] + + +def test_a_mixed_round_reports_both_outcomes_in_order() -> None: + """A search that worked followed by a fetch that was blocked is two facts.""" + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Partial answer.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_search_tool_result", SEARCH_OK), + block_start("server_tool_use"), + block_stop("web_fetch_tool_result", FETCH_BLOCKED), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.statuses == [ + {"content": "Searched the web", "steps": None, "summary": None}, + {"content": "Skipped a page outside the allowed sources", "steps": None, "summary": None}, + ] + + +def test_a_result_block_with_no_content_claims_nothing() -> None: + """Better to say nothing than to claim a lookup we cannot see the outcome of.""" + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Answer.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_fetch_tool_result", None), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.statuses == [] + assert planner._segments == [] + + +def test_a_successful_fetch_says_it_read_the_page() -> None: + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Per the page.")]) + events = [ + block_start("server_tool_use"), + block_stop("web_fetch_tool_result", FETCH_OK), + ] + planner.client = FakeClient(FakeStream(events, final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.statuses == [ + {"content": "Read a page from the web", "steps": None, "summary": None}, + ] + + +def test_the_spinner_uses_the_shared_web_status_pool() -> None: + planner = make_run_planner() + planner.client = FakeClient(FakeStream([block_start("server_tool_use")], FakeResponse("end_turn", []))) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.thinking == [STATUS_SEARCHING_WEB] + + +def test_text_deltas_still_stream_alongside_the_new_branches() -> None: + planner = make_run_planner() + final = FakeResponse("end_turn", [FakeTextBlock("Hi there")]) + planner.client = FakeClient(FakeStream([text_delta("Hi "), text_delta("there")], final)) + manager = StubStreamManager() + + planner._call_api([], [], True, manager) + + assert manager.text == ["Hi ", "there"] + + +class FakeSearchResult: + """A successful web_search_tool_result: content is a list.""" + + type = "web_search_tool_result" + + def __init__(self) -> None: + self.content = [{"type": "web_search_result", "url": "https://docs.dhis2.org/a"}] + + +class FakeSearchError: + """A failed web_search_tool_result: content is a bare object.""" + + type = "web_search_tool_result" + + def __init__(self) -> None: + self.content = {"type": "web_search_tool_result_error", "error_code": "max_uses_exceeded"} + + +def test_meta_counts_searches_and_fetch_hostnames() -> None: + planner = make_run_planner() + planner.web_search_enabled = True + responses = [ + FakeResponse("end_turn", [ + FakeServerToolUse("web_search", {"query": "dhis2 tracker"}, block_id="s1"), + FakeSearchResult(), + FakeServerToolUse("web_fetch", {"url": "https://docs.dhis2.org/en/tracker.html?q=secret"}, block_id="f1"), + FakeServerToolUse("web_fetch", {"url": "https://docs.dhis2.org/en/events.html"}, block_id="f2"), + FakeTextBlock("Here is the shape."), + ]), + ] + + result = run_with(planner, responses) + + assert (result.meta["web_searches"], result.meta["web_fetches"]) == (1, 2) + # Hostnames only, deduplicated, and the ?q=secret query string is dropped. + assert result.meta["web_domains"] == ["docs.dhis2.org"] + assert result.meta["web_search_downgraded"] is False + + +def test_meta_counting_survives_the_error_result_shape() -> None: + """A failed search returns content as an object. Counting + reads server_tool_use blocks and must not touch either shape.""" + planner = make_run_planner() + planner.web_search_enabled = True + responses = [ + FakeResponse("end_turn", [ + FakeServerToolUse("web_search", {"query": "dhis2"}, block_id="s1"), + FakeSearchError(), + FakeTextBlock("Could not find it."), + ]), + ] + + result = run_with(planner, responses) + + assert result.meta["web_searches"] == 1 + assert result.meta["web_fetches"] == 0 + assert result.meta["web_domains"] == [] + + +def test_meta_sums_web_usage_across_rounds() -> None: + """A paused search continues into a second round; counts must accumulate.""" + planner = make_run_planner() + planner.web_search_enabled = True + responses = [ + FakeResponse("pause_turn", [ + FakeServerToolUse("web_search", {"query": "dhis2"}, block_id="s1"), + FakeServerToolUse("web_fetch", {"url": "https://docs.dhis2.org/a.html"}, block_id="f1"), + ]), + FakeResponse("end_turn", [ + FakeServerToolUse("web_fetch", {"url": "https://docs.dhis2.org/b.html"}, block_id="f2"), + FakeServerToolUse("web_fetch", {"url": "https://docs.openfn.org/c.html"}, block_id="f3"), + FakeTextBlock("Done."), + ]), + ] + + result = run_with(planner, responses) + + assert (result.meta["web_searches"], result.meta["web_fetches"]) == (1, 3) + assert result.meta["web_domains"] == ["docs.dhis2.org", "docs.openfn.org"] + + +def test_meta_omits_the_web_fields_when_web_search_is_off() -> None: + planner = make_run_planner() + responses = [FakeResponse("end_turn", [FakeTextBlock("Plain answer.")])] + + result = run_with(planner, responses) + + assert "web_searches" not in result.meta + assert "web_search_downgraded" not in result.meta + + +def make_bad_request(message: str = "web search is not enabled for this account") -> BadRequestError: + request = httpx.Request("POST", "https://api.anthropic.com/v1/messages") + body = {"type": "error", "error": {"type": "invalid_request_error", "message": message}} + return BadRequestError(message, response=httpx.Response(400, request=request, json=body), body=body) + + +def test_a_bad_request_with_web_tools_retries_without_them() -> None: + planner = make_run_planner() + planner.web_tools = build_web_tools(WEB_CONFIG) + planner.tools = TOOL_DEFINITIONS + planner.web_tools + planner.web_search_enabled = True + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + tools_per_call = [] + + def fail_then_answer(_system: object, _messages: object, _stream: object, _manager: object, **_kwargs: object) -> FakeResponse: + tools_per_call.append([t.get("name") for t in planner.tools]) + if len(tools_per_call) == 1: + raise make_bad_request() + return FakeResponse("end_turn", [FakeTextBlock("Answered without the web.")]) + + with patch.object(PlannerAgent, "_call_api", side_effect=fail_then_answer): + result = planner.run("q", None, None, [], stream=False) + + # Two calls, first with the web tools, the retry without. + assert [("web_search" in names) for names in tools_per_call] == [True, False] + assert result.response == "Answered without the web." + assert result.meta["web_search_downgraded"] is True + assert result.response_segments[0] == { + "type": "status", + "content": "Web search is unavailable for this account — answering without it", + } + + +def test_the_downgrade_rebuilds_the_system_prompt_without_the_web_block() -> None: + """Otherwise the prompt still advertises two tools that are no longer sent.""" + planner = make_run_planner() + planner.web_tools = build_web_tools(WEB_CONFIG) + planner.tools = TOOL_DEFINITIONS + planner.web_tools + planner.web_search_enabled = True + planner.config_loader = StubPromptLoader( + planner_system_prompt="BASE PROMPT", + planner_web_tools_prompt="WEB ADDENDUM {domains}", + ) + systems = [] + + def fail_then_answer(system: list, _messages: object, _stream: object, _manager: object, **_kwargs: object) -> FakeResponse: + systems.append([block["text"] for block in system]) + if len(systems) == 1: + raise make_bad_request() + return FakeResponse("end_turn", [FakeTextBlock("Answered without the web.")]) + + with patch.object(PlannerAgent, "_call_api", side_effect=fail_then_answer): + planner.run("q", None, None, [], stream=False) + + # The first call carries the addendum, and the retry should not. + assert systems == [["BASE PROMPT", "WEB ADDENDUM docs.dhis2.org"], ["BASE PROMPT"]] + + +def test_a_bad_request_without_web_tools_is_not_retried() -> None: + planner = make_run_planner() + calls = [] + + def always_fail(*_args: object, **_kwargs: object) -> FakeResponse: + calls.append(1) + raise make_bad_request("prompt is too long") + + with patch.object(PlannerAgent, "_build_system_prompt", return_value=[]), \ + patch.object(PlannerAgent, "_call_api", side_effect=always_fail), \ + pytest.raises(ApolloError): + planner.run("q", None, None, [], stream=False) + + assert calls == [1] + + +def test_an_unrelated_bad_request_is_not_blamed_on_web_search() -> None: + """A 400 the web tools did not cause must not produce the web-search status.""" + planner = make_run_planner() + planner.web_tools = build_web_tools(WEB_CONFIG) + planner.tools = TOOL_DEFINITIONS + planner.web_tools + planner.web_search_enabled = True + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + + def always_fail(*_args: object, **_kwargs: object) -> FakeResponse: + raise make_bad_request("prompt is too long: 250000 tokens > 200000 maximum") + + with patch.object(PlannerAgent, "_call_api", side_effect=always_fail), \ + pytest.raises(ApolloError) as excinfo: + planner.run("q", None, None, [], stream=False) + + # The retry failed too, so the web tools were not the cause. The real + # error reaches details (not the user-facing message) and nothing is + # recorded about web search. + assert "prompt is too long" in excinfo.value.details["upstream_message"] + assert planner._segments == [] + + +def test_the_web_search_status_is_only_sent_once_the_retry_has_earned_it() -> None: + planner = make_run_planner() + planner.web_tools = build_web_tools(WEB_CONFIG) + planner.tools = TOOL_DEFINITIONS + planner.web_tools + planner.web_search_enabled = True + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + segments_at_each_call = [] + + def fail_then_answer(*_args: object, **_kwargs: object) -> FakeResponse: + # Snapshot before the retry runs: nothing may have been claimed yet. + segments_at_each_call.append(list(planner._segments)) + if len(segments_at_each_call) == 1: + raise make_bad_request() + return FakeResponse("end_turn", [FakeTextBlock("Answered without the web.")]) + + with patch.object(PlannerAgent, "_call_api", side_effect=fail_then_answer): + result = planner.run("q", None, None, [], stream=False) + + # Neither the first call nor the retry saw a status already recorded. + assert segments_at_each_call == [[], []] + assert result.response_segments[0]["type"] == "status" + + +def test_a_second_bad_request_surfaces_the_original_error() -> None: + planner = make_run_planner() + planner.web_tools = build_web_tools(WEB_CONFIG) + planner.tools = TOOL_DEFINITIONS + planner.web_tools + planner.web_search_enabled = True + planner.config_loader = StubPromptLoader(planner_system_prompt="BASE PROMPT") + errors = [make_bad_request("web search is not enabled"), make_bad_request("something else")] + + def always_fail(*_args: object, **_kwargs: object) -> FakeResponse: + raise errors.pop(0) + + with patch.object(PlannerAgent, "_call_api", side_effect=always_fail), \ + pytest.raises(ApolloError) as excinfo: + planner.run("q", None, None, [], stream=False) + + assert errors == [] + assert "web search is not enabled" in excinfo.value.details["upstream_message"] + assert "something else" not in excinfo.value.details["upstream_message"] + # The retry failed, so nothing was claimed about web search. + assert planner._segments == [] diff --git a/services/global_chat/tests/unit/test_router.py b/services/global_chat/tests/unit/test_router.py index 152ce256..76b5881e 100644 --- a/services/global_chat/tests/unit/test_router.py +++ b/services/global_chat/tests/unit/test_router.py @@ -2,6 +2,7 @@ from unittest.mock import patch +from global_chat.planner import PlannerResult from global_chat.router import RouterAgent, RouterDecision, RouterResult from yaml_utils import workflow_has_job_code @@ -29,6 +30,7 @@ def make_router() -> RouterAgent: """Build a RouterAgent without config or an Anthropic client.""" router = RouterAgent.__new__(RouterAgent) router.api_key = "test-key" + router.config_loader = None router.routing_usage = {} router._input_attachments = [] router._user = None @@ -200,6 +202,39 @@ def test_low_confidence_direct_route_goes_to_planner() -> None: assert result.response == "planner answer" +def make_planner_agent_result() -> PlannerResult: + return PlannerResult( + response="planner answer", + response_segments=[{"type": "text", "content": "planner answer"}], + attachments=[], + history=[], + usage={"input_tokens": 10}, + meta={"agents": ["router", "planner"]}, + ) + + +def test_route_and_execute_defaults_web_search_off() -> None: + router = make_router() + decision = RouterDecision(destination="planner", confidence=5) + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()): + router.route_and_execute("build me a workflow", None, None, [], False) + + assert router._web_search is False + + +def test_planner_route_forwards_the_web_search_flag() -> None: + router = make_router() + router._web_search = True + + with patch("global_chat.planner.PlannerAgent") as planner_cls: + planner_cls.return_value.run.return_value = make_planner_agent_result() + router._route_to_planner("look up the DHIS2 tracker API", None, None, [], False, 5) + + assert planner_cls.call_args.kwargs["web_search"] is True + + def test_confident_direct_route_is_not_gated() -> None: router = make_router() decision = RouterDecision(destination="job_code_agent", confidence=3, job_key="fetch-patients") @@ -220,3 +255,34 @@ def test_confident_direct_route_is_not_gated() -> None: job_mock.assert_called_once() planner_mock.assert_not_called() assert result.response == "job answer" + + +def test_meta_marks_a_request_that_asked_for_web_search() -> None: + """Visible even when the request never reached the planner.""" + router = make_router() + decision = RouterDecision(destination="job_code_agent", confidence=5, job_key="fetch-patients") + job_result = RouterResult( + response="job answer", + response_segments=[], + attachments=[], + history=[], + usage={}, + meta={"agents": ["router", "job_code_agent"]}, + ) + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_job_chat", return_value=job_result): + result = router.route_and_execute("edit this", WORKFLOW_YAML, None, [], False, web_search=True) + + assert result.meta["web_search_requested"] is True + + +def test_meta_omits_web_search_requested_by_default() -> None: + router = make_router() + decision = RouterDecision(destination="planner", confidence=5) + + with patch.object(RouterAgent, "_make_routing_decision", return_value=decision), \ + patch.object(RouterAgent, "_route_to_planner", return_value=make_planner_result()): + result = router.route_and_execute("build me a workflow", None, None, [], False) + + assert "web_search_requested" not in result.meta diff --git a/services/global_chat/tests/unit/test_tool_definitions.py b/services/global_chat/tests/unit/test_tool_definitions.py new file mode 100644 index 00000000..fd4577c8 --- /dev/null +++ b/services/global_chat/tests/unit/test_tool_definitions.py @@ -0,0 +1,65 @@ +"""Unit tests for the planner's web tool definitions.""" + +from global_chat.tools.tool_definitions import build_web_tools + +MAX_USES = 3 +MAX_CONTENT_TOKENS = 7500 +ALLOWED_DOMAINS = ["docs.dhis2.org", "docs.openfn.org"] + +WEB_CONFIG = { + "planner": { + "web_search": { + "max_uses": MAX_USES, + "max_content_tokens": MAX_CONTENT_TOKENS, + "allowed_domains": ALLOWED_DOMAINS, + }, + }, +} + + +def by_name(tools: list[dict]) -> dict[str, dict]: + return {tool["name"]: tool for tool in tools} + + +def test_no_web_search_block_leaves_the_tools_off() -> None: + assert build_web_tools({"planner": {}}) == [] + assert build_web_tools({}) == [] + + +def test_empty_allowlist_leaves_the_tools_off() -> None: + """The empty allowlist is the server-side kill switch, not open-web.""" + assert build_web_tools({"planner": {"web_search": {"allowed_domains": []}}}) == [] + assert build_web_tools({"planner": {"web_search": {"max_uses": 5}}}) == [] + + +def test_populated_allowlist_builds_both_tools_with_the_current_type_strings() -> None: + tools = by_name(build_web_tools(WEB_CONFIG)) + + assert tools["web_search"]["type"] == "web_search_20260209" + assert tools["web_fetch"]["type"] == "web_fetch_20260209" + + +def test_max_content_tokens_reaches_the_fetch_tool_only() -> None: + """Sending max_content_tokens on web_search is a 400 — it must not be set.""" + tools = by_name(build_web_tools(WEB_CONFIG)) + + assert "max_content_tokens" not in tools["web_search"] + assert tools["web_fetch"]["max_content_tokens"] == MAX_CONTENT_TOKENS + + +def test_allowlist_and_max_uses_go_on_both_tools() -> None: + tools = by_name(build_web_tools(WEB_CONFIG)) + + for tool in tools.values(): + assert tool["max_uses"] == MAX_USES + assert tool["allowed_domains"] == ALLOWED_DOMAINS + assert "blocked_domains" not in tool + + +def test_the_configured_allowlist_is_copied_not_aliased() -> None: + """A caller mutating the returned list must not edit config in place.""" + config = {"planner": {"web_search": {"allowed_domains": ["docs.dhis2.org"]}}} + + build_web_tools(config)[0]["allowed_domains"].append("evil.example") + + assert config["planner"]["web_search"]["allowed_domains"] == ["docs.dhis2.org"] diff --git a/services/global_chat/tools/tool_definitions.py b/services/global_chat/tools/tool_definitions.py index 8ad98f58..5ed3f814 100644 --- a/services/global_chat/tools/tool_definitions.py +++ b/services/global_chat/tools/tool_definitions.py @@ -94,3 +94,34 @@ CALL_JOB_CODE_AGENT_TOOL, INSPECT_JOB_CODE_TOOL ] + + +def build_web_tools(config: dict) -> list[dict]: + """Build Anthropic's server-side web search and fetch tool definitions. + + `allowed_domains` is passed through from config unvalidated and is the boundary + between fetched page content and the planner. An empty list returns no tools, + which is the kill switch for the feature. + """ + web_config = (config.get("planner") or {}).get("web_search") or {} + allowed_domains = list(web_config.get("allowed_domains") or []) + if not allowed_domains: + return [] + + max_uses = web_config.get("max_uses", 5) + + return [ + { + "type": "web_search_20260209", + "name": "web_search", + "max_uses": max_uses, + "allowed_domains": allowed_domains, + }, + { + "type": "web_fetch_20260209", + "name": "web_fetch", + "max_uses": max_uses, + "max_content_tokens": web_config.get("max_content_tokens", 10000), + "allowed_domains": list(allowed_domains), + }, + ] diff --git a/services/streaming_util.py b/services/streaming_util.py index 4e2465c6..fdcb6d9c 100644 --- a/services/streaming_util.py +++ b/services/streaming_util.py @@ -84,6 +84,10 @@ "Writing code...", ] +STATUS_SEARCHING_WEB = [ + "Searching the web...", +] + @dataclass class ContentBlock: