"""Behavior tests for the connector leg of the tool_search bridge. DI-callable idiom (test_managed_tool_gateway.py precedent): remote legs are injected as plain callables; no module mocks, no network. Local-only behavior is pinned byte-identical when the remote leg fails (D32). """ import json import logging import pytest from agent.tool_dispatch_helpers import _peel_bridge_call from tools.connectors.gateway.bridge import ConnectorLeg, connector_describe from tools.connectors.gateway.errors import GatewayAuthError from tools.tool_search import ( CONNECTOR_BATCH_SENTINEL, ToolSearchConfig, assemble_tool_defs, dispatch_tool_describe, dispatch_tool_search, resolve_underlying_call, ) from tools.tool_search_validation import normalize_tool_call_entries def _tool_search_description(tool_defs): # session_search is in the default defer set, so the bridge activates in # both arms for the same reason: a deferrable local tool exists. defs = tool_defs + [{"type": "function", "function": { "name": "session_search", "description": "Search past sessions", "parameters": {}}}] assembled = assemble_tool_defs( defs, context_length=200_000, config=ToolSearchConfig.from_raw({"enabled": "on"})) assert assembled.activated return next(td["function"]["description"] for td in assembled.tool_defs if td["function"]["name"] == "tool_search") def test_tool_search_names_manage_connections_only_when_the_session_has_it(): """The model learns that connectors__ names belong to accounts managed by manage_connections from the tool_search description, but only when that tool is in the session. Signed out (or connectors off) the tool is absent and the description must not name a tool the model cannot call.""" with_connections = _tool_search_description(_local_defs()) assert "manage_connections" in with_connections assert "connectors__" in with_connections without = _tool_search_description( [td for td in _local_defs() if td["function"]["name"] != "manage_connections"]) assert "manage_connections" not in without def _local_defs(): """One deferrable (mcp-toolset) tool, one core-shaped tool.""" return [ {"type": "function", "function": {"name": "manage_connections", "parameters": {}}}, { "type": "function", "function": { "name": "mcp__github__create_issue", "description": "Create a GitHub issue", "parameters": {"type": "object", "properties": {}, "required": ["title"]}, }, }, ] # --------------------------------------------------------------------------- # resolve_underlying_call: batch shapes # --------------------------------------------------------------------------- def test_resolve_single_connector_entry_returns_sentinel(): name, args, err = resolve_underlying_call( {"calls": [{"name": "connectors__gmail__SEND_EMAIL", "arguments": {"to": "x"}}]} ) assert err is None assert name == CONNECTOR_BATCH_SENTINEL assert args["calls"][0]["name"] == "connectors__gmail__SEND_EMAIL" assert args["calls"][0]["arguments"] == {"to": "x"} def test_resolve_multi_local_batch_echoes_first_entry_as_the_retry_shape(): # The correction restates the valid single-entry shape with the caller's OWN first # entry (a 9B model re-sent the identical two-entry array when told only the rule). first = {"name": "some_local_tool", "arguments": {"query": "alpha", "limit": 20}} name, args, err = resolve_underlying_call({"calls": [first, {"name": "another_local", "arguments": {}}]}) assert name is None assert "you sent 2" in err retry = err.split("Retry with only: ", 1)[1].split(" then issue", 1)[0] assert json.loads(retry) == {"calls": [first]} def test_resolve_unknown_name_points_at_tool_search_not_direct_call(): # An unregistered name (typically a deferred MCP tool cited by its bare suffix) must # NOT be told "call it directly" — that is the opposite of the required correction. from tools.registry import registry name, args, err = resolve_underlying_call({"name": "not_a_real_tool", "arguments": {}}) assert name is None assert "not a known tool name" in err and "call it directly" not in err.lower() registry.register(name="mcp__mempalace__mempalace_search", toolset="mcp-mempalace", handler=lambda a, **kw: "{}", schema={"name": "mcp__mempalace__mempalace_search", "description": "x", "parameters": {"type": "object", "properties": {}}}) _, _, err = resolve_underlying_call({"name": "mempalace_search", "arguments": {}}) registry.deregister("mcp__mempalace__mempalace_search") assert "Did you mean 'mcp__mempalace__mempalace_search'?" in err _, _, err = resolve_underlying_call({"name": "read_file", "arguments": {}}) assert "directly-listed tool" in err and "not a known tool" not in err def test_resolve_legacy_connector_single_shape_routes_to_sentinel(): name, args, err = resolve_underlying_call( {"name": "connectors__gmail__CREATE_EMAIL_DRAFT", "arguments": {}} ) assert err is None assert name == CONNECTOR_BATCH_SENTINEL assert len(args["calls"]) == 1 def test_normalize_parses_string_envelope_batch(): """#114484: a model-emitted JSON-string batch envelope parses like the array form.""" calls = [{"name": "session_search", "arguments": {"query": "x"}}] entries, err = normalize_tool_call_entries({"calls": json.dumps(calls)}) assert err is None assert entries == calls def test_normalize_parses_string_envelope_single_dict(): """#114484: a stringified single dict normalizes to a batch of one.""" entries, err = normalize_tool_call_entries( {"calls": json.dumps({"name": "session_search", "arguments": {"query": "x"}})} ) assert err is None assert entries == [{"name": "session_search", "arguments": {"query": "x"}}] @pytest.mark.parametrize( "bad,expected_fragment", [ ({}, "requires 'calls'"), ({"calls": []}, "non-empty array"), ({"calls": [{"arguments": {}}]}, "requires a 'name'"), ({"calls": [{"name": "tool_search"}]}, "itself a bridge tool"), ({"calls": [{"name": "x", "arguments": "not json {"}]}, "not valid JSON"), ({"calls": [{"name": "x", "arguments": 42}]}, "must be an object"), ({"calls": "nope"}, "not valid JSON"), ({"calls": json.dumps({"query": "x"})}, "requires a 'name'"), ], ) def test_normalize_rejects_malformed_batches(bad, expected_fragment): entries, err = normalize_tool_call_entries(bad) assert entries == [] assert expected_fragment in (err or "") # --------------------------------------------------------------------------- # dispatch_tool_search: remote merge # --------------------------------------------------------------------------- def _fake_connector_search(queries): assert queries == [{"use_case": "send an email"}] return ConnectorLeg(payload={ "results": [ {"index": 1, "use_case": "send an email", "tools": ["GMAIL_SEND_EMAIL", "ORPHAN_TOOL"]}, ], "schemas": { "GMAIL_SEND_EMAIL": { "connector": "gmail", "tool": "GMAIL_SEND_EMAIL", "description": "Send an email via gmail", "input_schema": {"type": "object", "required": ["to", "subject"]}, }, # ORPHAN_TOOL deliberately has no schema entry: without a # connector it cannot compose a callable name and must be dropped. }, "connections": [{"connector": "gmail", "connected": False, "description": ""}], }) def _registered_local_defs(): """Deferrable local tools the registry knows, so they enter the BM25 catalog: an issue tracker whose descriptions mention email notifications, and an unrelated tool.""" from tools.registry import registry specs = [ ("mcp__tracker__create_issue", "Create an issue. Sends an email notification to the team."), ("mcp__tracker__list_issues", "List issues in a project. Email digests are optional."), ("mcp__tracker__archive_project", "Archive a project and its issues."), ] defs = [] for name, desc in specs: schema = {"name": name, "description": desc, "parameters": {"type": "object", "properties": {"id": {"type": "string"}}}} registry.register(name=name, handler=lambda a, **k: "{}", schema=schema, toolset="mcp-tracker") defs.append({"type": "function", "function": schema}) return [{"type": "function", "function": {"name": "manage_connections", "parameters": {}}}] + defs, [n for n, _ in specs] def test_connector_intent_is_not_starved_by_local_tools_sharing_one_word(): """The reported bug: with a large local catalog, tools that merely shared 'email' filled every slot and the gmail connector tool never appeared. Ranked as one corpus with the rarest-token gate ('gmail' is in one document), the connector tool is the only result.""" from tools.registry import registry defs, names = _registered_local_defs() try: out = json.loads(dispatch_tool_search( {"queries": ["send gmail email"], "limit": 5}, current_tool_defs=defs, connector_search=lambda q: ConnectorLeg(payload={ "results": [{"use_case": "send gmail email", "tools": ["GMAIL_SEND_EMAIL"]}], "schemas": {"GMAIL_SEND_EMAIL": { "connector": "gmail", "tool": "GMAIL_SEND_EMAIL", "description": "Send an email via gmail", "input_schema": {}}}, }))) assert out["results"][0]["matches"] == ["connectors__gmail__SEND_EMAIL"] finally: for n in names: registry.deregister(n) def test_both_sources_answer_within_one_limit(): """When a local MCP server and a connector both serve the same service, both surface, ranked by the same BM25 pass, and `limit` caps the group as a whole.""" from tools.registry import registry defs, names = _registered_local_defs() try: out = json.loads(dispatch_tool_search( {"queries": ["tracker create issue"], "limit": 2}, current_tool_defs=defs, connector_search=lambda q: ConnectorLeg(payload={ "results": [{"use_case": "tracker create issue", "tools": ["TRACKER_CREATE_ISSUE"]}], "schemas": {"TRACKER_CREATE_ISSUE": { "connector": "tracker", "tool": "TRACKER_CREATE_ISSUE", "description": "Create a tracker issue", "input_schema": {}}}, }))) matches = out["results"][0]["matches"] assert len(matches) == 2 assert set(matches) == {"mcp__tracker__create_issue", "connectors__tracker__CREATE_ISSUE"} finally: for n in names: registry.deregister(n) def test_search_composes_lowercase_connector_from_vendor_cased_schema(): # The gateway search surface leaks vendor-cased connector slugs for # custom toolkits; the composed name must carry the lowercase catalog # form or the gateway's own policy gates refuse the call. def cased_search(queries): return ConnectorLeg(payload={ "results": [{"index": 1, "tools": ["CUSTOM_X_READ"]}], "schemas": { "CUSTOM_X_READ": { "connector": "CUSTOM_X", "tool": "CUSTOM_X_READ", "description": "d", "input_schema": {}, } }, }) out = json.loads( dispatch_tool_search( {"queries": ["custom_x read"]}, current_tool_defs=_local_defs(), connector_search=cased_search, ) ) assert "connectors__custom_x__READ" in out["results"][0]["matches"] def test_search_merges_remote_hits_tagged_as_connectors(): out = json.loads( dispatch_tool_search( {"queries": ["send an email"]}, current_tool_defs=_local_defs(), connector_search=_fake_connector_search, ) ) matches = out["results"][0]["matches"] composed = "connectors__gmail__SEND_EMAIL" assert composed in matches assert all("ORPHAN_TOOL" not in m for m in matches) record = out["tools"][composed] assert record["source"] == "connectors" assert record["source_name"] == "gmail" assert record["required"] == ["to", "subject"] @pytest.mark.parametrize("order", [("GMAIL_FETCH_PROFILE", "FETCH_PROFILE"), ("FETCH_PROFILE", "GMAIL_FETCH_PROFILE")]) def test_search_keeps_only_the_twin_a_colliding_name_reaches(order, caplog): """Composition is not injective: GMAIL_FETCH_PROFILE and a literal FETCH_PROFILE on gmail both compose to connectors__gmail__FETCH_PROFILE, and describe/execute decode that name to GMAIL_FETCH_PROFILE. If a vendor ever ships both, search must not describe the literal under a name that runs the prefixed tool, whichever the gateway listed first, and must say so in the log rather than alias silently.""" def twins(queries): return ConnectorLeg(payload={ "results": [{"index": 1, "tools": list(order)}], "schemas": { "GMAIL_FETCH_PROFILE": {"connector": "gmail", "tool": "GMAIL_FETCH_PROFILE", "description": "prefixed twin", "input_schema": {}}, "FETCH_PROFILE": {"connector": "gmail", "tool": "FETCH_PROFILE", "description": "literal twin", "input_schema": {}}, }, }) with caplog.at_level(logging.WARNING, logger="tools.connectors.search"): out = json.loads(dispatch_tool_search( {"queries": ["gmail fetch profile"]}, current_tool_defs=_local_defs(), connector_search=twins, )) assert out["results"][0]["matches"] == ["connectors__gmail__FETCH_PROFILE"] assert out["tools"]["connectors__gmail__FETCH_PROFILE"]["description"] == "prefixed twin" warnings = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING] assert len(warnings) == 1 assert "GMAIL_FETCH_PROFILE" in warnings[0] and "FETCH_PROFILE" in warnings[0] def test_search_limit_caps_the_group_across_both_legs_and_counts_total(): def many_hits(queries): slugs = [f"CUSTOM_X_TOOL_{i}" for i in range(9)] return ConnectorLeg(payload={ "results": [{"index": 1, "tools": slugs}], "schemas": { s: {"connector": "custom_x", "tool": s, "description": "widget", "input_schema": {}} for s in slugs }, }) out = json.loads( dispatch_tool_search( {"queries": ["custom_x widget"], "limit": 3}, current_tool_defs=_local_defs(), connector_search=many_hits, ) ) matches = out["results"][0]["matches"] assert len(matches) == 3 # limit is the per-query cap across BOTH legs assert all(m.startswith("connectors__") for m in matches) # total_available counts returned remote tools on top of the local catalog # (empty here: the fake def is not registry-backed in this test env). assert out["total_available"] == 3 def test_search_drops_remote_group_with_mismatched_use_case_echo(): def misaligned(queries): return ConnectorLeg(payload={ "results": [{"index": 1, "use_case": "SOMETHING ELSE", "tools": ["CUSTOM_X_READ"]}], "schemas": { "CUSTOM_X_READ": {"connector": "custom_x", "tool": "CUSTOM_X_READ", "description": "d", "input_schema": {}} }, }) out = json.loads( dispatch_tool_search( {"queries": ["send an email"]}, current_tool_defs=_local_defs(), connector_search=misaligned, ) ) assert not any(m.startswith("connectors__") for m in out["results"][0]["matches"]) def test_search_keeps_local_results_and_names_the_hosted_failure(): def exploding_search(queries): raise RuntimeError("gateway exploded") local_only = json.loads(dispatch_tool_search( {"queries": ["send an email"]}, current_tool_defs=_local_defs(), connector_search=lambda queries: ConnectorLeg(), )) with_failure = json.loads(dispatch_tool_search( {"queries": ["send an email"]}, current_tool_defs=_local_defs(), connector_search=exploding_search, )) assert "connectors" not in local_only connectors = with_failure.pop("connectors") assert local_only == with_failure assert connectors["status"] == "unavailable" and connectors["reason"] == "unreachable" def test_search_never_sends_the_gateway_more_use_cases_than_it_accepts(): """The gateway's search route returns HTTP 502 above 7 use_cases per request, and one tool_search call maps to one gateway request. Seven queries reach it in one request; eight are refused before any request is made, so the model gets a retry hint and the gateway never sees a request it cannot answer.""" sent = [] def recording_search(use_cases): sent.append(use_cases) return ConnectorLeg() seven = [f"query {i}" for i in range(7)] parsed = json.loads(dispatch_tool_search( {"queries": seven}, current_tool_defs=_local_defs(), connector_search=recording_search)) assert "error" not in parsed assert sent == [[{"use_case": q} for q in seven]] sent.clear() parsed = json.loads(dispatch_tool_search( {"queries": seven + ["query 7"]}, current_tool_defs=_local_defs(), connector_search=recording_search)) assert "too many queries" in parsed["error"] assert sent == [] # --------------------------------------------------------------------------- # dispatch_tool_describe: remote merge # --------------------------------------------------------------------------- def test_describe_merges_remote_schema_and_leaves_misses_in_not_found(): composed = "connectors__gmail__SEND_EMAIL" stale = "connectors__gmail__GONE_TOOL" def fake_describe(names): assert set(names) == {composed, stale} return ConnectorLeg(payload={ "tools": {composed: {"description": "Send an email", "parameters": {"type": "object"}}}}) out = json.loads( dispatch_tool_describe( {"names": [composed, stale]}, current_tool_defs=_local_defs(), connector_describe=fake_describe, ) ) assert out["tools"][composed]["parameters"] == {"type": "object"} assert stale in out["not_found"] assert "errors" not in out # a connector miss is stale/unknown, not an error def test_describe_connector_names_fall_to_not_found_when_dark_and_to_connectors_when_the_leg_fails(): composed = "connectors__gmail__SEND_EMAIL" out = json.loads( dispatch_tool_describe( {"names": [composed]}, current_tool_defs=_local_defs(), connector_describe=lambda names: ConnectorLeg(), ) ) assert out["not_found"] == [composed] def exploding_describe(names): raise RuntimeError("gateway exploded") failed = json.loads( dispatch_tool_describe( {"names": [composed]}, current_tool_defs=_local_defs(), connector_describe=exploding_describe, ) ) assert "not_found" not in failed and "hint" not in failed assert failed["connectors"]["status"] == "unavailable" assert failed["connectors"]["reason"] == "unreachable" assert failed["connectors"]["names"] == [composed] # --------------------------------------------------------------------------- # planner admission: only PURE connector batches are parallel-safe # --------------------------------------------------------------------------- def test_peel_admits_pure_connector_batch_as_sentinel(): name, args = _peel_bridge_call( "tool_call", {"calls": [ {"name": "connectors__gmail__SEND_EMAIL", "arguments": {}}, {"name": "connectors__slack__POST_MESSAGE", "arguments": {}}, ]}, ) assert name == CONNECTOR_BATCH_SENTINEL def test_peel_keeps_mixed_and_local_batches_as_sequential_barrier(): mixed = {"calls": [ {"name": "connectors__gmail__SEND_EMAIL", "arguments": {}}, {"name": "write_file", "arguments": {"path": "x"}}, ]} name, args = _peel_bridge_call("tool_call", mixed) assert name == "tool_call" # barrier: local entries never got admission all_local = {"calls": [ {"name": "write_file", "arguments": {"path": "x"}}, {"name": "read_file", "arguments": {"path": "x"}}, ]} name, _ = _peel_bridge_call("tool_call", all_local) assert name == "tool_call" # --------------------------------------------------------------------------- # run_remote through production dispatch: vendor-slug restoration, the # one-pass literal fallback, and the gateway request body # # tool_call batches re-enter core dispatch once per connector entry, so each # entry reaches run_remote alone. The gateway is swapped at the bridge's # client factory, the same seam the real client is created through. # --------------------------------------------------------------------------- def _connectors_on(monkeypatch, client_factory): from tools.registry import invalidate_check_fn_cache from tools.connectors.gateway import bridge, config monkeypatch.setattr(config, "connectors_available", lambda: True) monkeypatch.setattr(bridge, "connectors_available", lambda: True) monkeypatch.setattr(bridge, "_default_client_factory", client_factory) invalidate_check_fn_cache() def _tool_call(calls): import model_tools return json.loads(model_tools.handle_function_call( "tool_call", {"calls": calls}, enabled_toolsets=["connections"], session_id="bridge-session", skip_pre_tool_call_hook=True, skip_tool_request_middleware=True, skip_tool_execution_middleware=True)) def test_execute_falls_back_once_for_literal_slug_without_touching_siblings(monkeypatch): class FakeClient: def __init__(self): self.calls = [] def execute(self, planned): self.calls.append([plan.tool for plan in planned]) (plan,) = planned if plan.tool == "GRANOLA_FETCH_NOTES": return [{"data": None, "error": {"code": "TOOL_NOT_FOUND", "message": "missing"}}] if plan.tool == "SLACK_POST_MESSAGE": return [{"data": None, "error": {"code": "TOOL_NOT_ALLOWED", "message": "blocked"}}] return [{"data": f"ran {plan.tool}", "error": None}] client = FakeClient() _connectors_on(monkeypatch, lambda: client) out = _tool_call([ {"name": "connectors__gmail__SEND_EMAIL", "arguments": {}}, {"name": "connectors__granola__FETCH_NOTES", "arguments": {}}, {"name": "connectors__slack__POST_MESSAGE", "arguments": {}}, ]) # Every entry crosses the wire under its restored vendor slug. Only the # confirmed TOOL_NOT_FOUND miss is retried, once, under the literal slug; # the success and the TOOL_NOT_ALLOWED sibling are never re-sent. assert client.calls == [ ["GMAIL_SEND_EMAIL"], ["GRANOLA_FETCH_NOTES"], ["FETCH_NOTES"], ["SLACK_POST_MESSAGE"]] assert out["results"][0]["response"] == "ran GMAIL_SEND_EMAIL" assert out["results"][1] == { "index": 1, "name": "connectors__granola__FETCH_NOTES", "response": "ran FETCH_NOTES"} assert out["results"][2]["error"]["code"] == "TOOL_NOT_ALLOWED" assert out["success_count"] == 2 and out["error_count"] == 1 @pytest.mark.parametrize("fail_at", ["GRANOLA_FETCH_NOTES", "FETCH_NOTES"]) def test_execute_transport_failure_degrades_only_the_failing_entry(monkeypatch, fail_at): class FakeClient: def execute(self, planned): (plan,) = planned if plan.tool == fail_at: raise RuntimeError("transport failed") if plan.tool == "GRANOLA_FETCH_NOTES": return [{"data": None, "error": {"code": "TOOL_NOT_FOUND", "message": "missing"}}] return [{"data": "sibling", "error": None}] _connectors_on(monkeypatch, FakeClient) out = _tool_call([ {"name": "connectors__gmail__SEND_EMAIL", "arguments": {}}, {"name": "connectors__granola__FETCH_NOTES", "arguments": {}}, ]) # Whether the primary send or the literal retry blows up, only that entry # degrades to PROVIDER_ERROR; the sibling keeps its result. assert out["results"][0]["response"] == "sibling" assert out["results"][1]["error"]["code"] == "PROVIDER_ERROR" assert out["success_count"] == 1 and out["error_count"] == 1 class _FakeResponse: def __init__(self, body): self.status_code = 200 self._body = body self.text = json.dumps(body) def json(self): return self._body class _RecordingTransport: """Records every outgoing request; answers each with `data` echoes.""" def __init__(self): self.requests = [] def request(self, method, url, *, headers=None, json=None, timeout=None): self.requests.append({"method": method, "url": url, "json": json}) tools = (json or {}).get("tools") or [] results = [ {"index": i, "connector": t.get("connector"), "tool": t.get("tool"), "data": "ok"} for i, t in enumerate(tools) ] return _FakeResponse( { "results": results, "successCount": len(results), "errorCount": 0, "totalCount": len(results), } ) def _recording_client_factory(transport): from tools.connectors.gateway.client import ConnectorClient return lambda: ConnectorClient( transport=transport, endpoint_resolver=lambda: "https://tool-gateway.test", header_provider=lambda url: {"Authorization": "Bearer nous-token"}, ) def _sent_tools(transport): assert len(transport.requests) == 1 return transport.requests[0]["json"]["tools"] def test_hook_rewrite_and_restored_vendor_slug_reach_the_gateway_request_body(monkeypatch): import hermes_cli.plugins as plugins transport = _RecordingTransport() _connectors_on(monkeypatch, _recording_client_factory(transport)) # A pre_tool_call redaction pass: the secret must never leave the process. monkeypatch.setattr(plugins, "_dispatch_pre_tool_call_hooks", lambda name, args, **kw: (None, {**args, "body": "[REDACTED]"})) out = _tool_call([{"name": "connectors__gmail__SEND_EMAIL", "arguments": {"to": "x@example.com", "body": "sk-secret"}}]) assert _sent_tools(transport) == [ {"connector": "gmail", "tool": "GMAIL_SEND_EMAIL", "arguments": {"to": "x@example.com", "body": "[REDACTED]"}}, ] assert "sk-secret" not in json.dumps(transport.requests[0]["json"]) assert out["results"][0]["response"] == "ok" # correlation survives the rewrite # --------------------------------------------------------------------------- # bridge.connector_describe: composed-name round trip # --------------------------------------------------------------------------- def test_connector_describe_maps_slugs_back_to_composed_names(): class FakeClient: def schemas(self, slugs): assert slugs == ["GMAIL_SEND_EMAIL", "SEND_EMAIL"] return { "schemas": { "GMAIL_SEND_EMAIL": { "connector": "gmail", "tool": "GMAIL_SEND_EMAIL", "description": "Send an email", "input_schema": {"type": "object"}, } }, "not_found": [], } out = connector_describe( ["connectors__gmail__SEND_EMAIL", "connectors__broken"], availability=lambda: True, client_factory=lambda: FakeClient(), ) assert out.payload["tools"]["connectors__gmail__SEND_EMAIL"]["parameters"] == {"type": "object"} def test_connector_describe_colliding_candidate_slugs_resolve_per_name(): # connectors__first__SECOND_X nominates (FIRST_SECOND_X, SECOND_X) and # connectors__second__X nominates (SECOND_X, X): the shared SECOND_X must # not be claimed globally by whichever name came first. Each name takes # its own best-ranked resolved candidate — here the first name's prefixed # primary is unknown to the gateway, so BOTH names land on SECOND_X. class FakeClient: def schemas(self, slugs): assert slugs == ["FIRST_SECOND_X", "SECOND_X", "X"] return { "schemas": { "SECOND_X": { "connector": "second", "tool": "SECOND_X", "description": "Shared slug", "input_schema": {"type": "object"}, } }, "not_found": ["FIRST_SECOND_X", "X"], } out = connector_describe( ["connectors__first__SECOND_X", "connectors__second__X"], availability=lambda: True, client_factory=lambda: FakeClient(), ) assert set(out.payload["tools"]) == { "connectors__first__SECOND_X", "connectors__second__X", } def test_connector_describe_prefers_each_names_prefixed_candidate(): # When BOTH of a name's candidates resolve, the prefixed primary wins — # the literal is only the recovery lane for slugs the encoder never # stripped. class FakeClient: def schemas(self, slugs): assert slugs == ["GMAIL_X", "X"] return { "schemas": { "GMAIL_X": { "connector": "gmail", "tool": "GMAIL_X", "description": "prefixed", "input_schema": {"type": "object", "title": "prefixed"}, }, "X": { "connector": "gmail", "tool": "X", "description": "literal", "input_schema": {"type": "object", "title": "literal"}, }, }, "not_found": [], } out = connector_describe( ["connectors__gmail__X"], availability=lambda: True, client_factory=lambda: FakeClient(), ) assert out.payload["tools"]["connectors__gmail__X"]["description"] == "prefixed" def test_connector_describe_is_empty_on_unavailable_and_names_the_failure_reason(): off = connector_describe(["connectors__g__T"], availability=lambda: False) assert off == ConnectorLeg() def boom(): raise RuntimeError("boom") assert connector_describe( ["connectors__g__T"], availability=lambda: True, client_factory=boom ) == ConnectorLeg(failure="unreachable") def rejected(): raise GatewayAuthError("token rejected", code="UNAUTHORIZED", status=401) assert connector_describe( ["connectors__g__T"], availability=lambda: True, client_factory=rejected ) == ConnectorLeg(failure="sign_in_expired")