mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-04 14:12:58 +00:00
Alex
574f96341e
The backend import package is now docsgpt, the name it will carry on PyPI; application was far too generic to install into anyone's site-packages. git mv plus a mechanical rewrite of every import, dotted string and path reference: 734 Python files, the compose files, Dockerfile, workflows, docs, setup scripts, devcontainer, k8s manifests, vscode config, pytest and coverage config, .gitignore. Behaviour is unchanged. Kept for one release: - A top-level application package whose meta-path finder resolves application.x.y to the already-imported docsgpt.x.y object, so old imports and entry points (celery -A application.app.celery, uvicorn application.asgi:asgi_app) keep working with a FutureWarning. - Celery registers every application.* task name as an alias of its docsgpt.* task on start-up, so messages queued by the previous release still run. The redbeat key prefix moves to redbeat:docsgpt:v2: so schedule entries the previous release wrote are left unread instead of firing twice. The backend image builds from the repository root (docker build -f docsgpt/Dockerfile .) so it can ship the alias package; a root .dockerignore allow-lists docsgpt/ and application/ and keeps caches, local data, .env files, the sample index files and the Dockerfile out. Compose and the image workflows point at the new context.
1557 lines
56 KiB
Python
1557 lines
56 KiB
Python
"""Tests for ToolExecutor — tool discovery, preparation, and execution."""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from docsgpt.agents.tool_executor import ToolExecutor
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestToolExecutorInit:
|
|
|
|
def test_default_state(self):
|
|
executor = ToolExecutor()
|
|
assert executor.user_api_key is None
|
|
assert executor.user is None
|
|
assert executor.agent_id is None
|
|
assert executor.tool_calls == []
|
|
assert executor._loaded_tools == {}
|
|
assert executor.conversation_id is None
|
|
|
|
def test_init_with_params(self):
|
|
executor = ToolExecutor(
|
|
user_api_key="key",
|
|
user="alice",
|
|
decoded_token={"sub": "alice"},
|
|
agent_id="agent-1",
|
|
)
|
|
assert executor.user_api_key == "key"
|
|
assert executor.user == "alice"
|
|
assert executor.agent_id == "agent-1"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestToolExecutorGetTools:
|
|
|
|
@staticmethod
|
|
def _patch_conn(monkeypatch, pg_conn):
|
|
from contextlib import contextmanager
|
|
|
|
@contextmanager
|
|
def _use_pg_conn():
|
|
yield pg_conn
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.db_readonly", _use_pg_conn
|
|
)
|
|
|
|
def test_get_tools_uses_api_key_when_present(self, pg_conn, monkeypatch):
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
tool = UserToolsRepository(pg_conn).create(user_id="alice", name="tool1")
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="alice",
|
|
name="a",
|
|
status="active",
|
|
key="test_key",
|
|
tools=[str(tool["id"])],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user_api_key="test_key", user="alice")
|
|
tools = executor.get_tools()
|
|
assert isinstance(tools, dict)
|
|
# The tool id should appear as key; tool_data['id'] is set
|
|
assert str(tool["id"]) in tools
|
|
assert tools[str(tool["id"])]["id"] == tool["id"]
|
|
|
|
def test_api_key_agent_never_resolves_workflow_only_builtins(self, pg_conn, monkeypatch):
|
|
"""read_document is workflow-only: a chat/scheduled agent carrying its
|
|
builtin id must not get it; workflow nodes get it via allowed_tool_ids."""
|
|
from docsgpt.agents.default_tools import default_tool_id
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
|
|
read_doc_id = default_tool_id("read_document")
|
|
artifact_id = default_tool_id("artifact_generator")
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="alice",
|
|
name="a",
|
|
status="active",
|
|
key="wf_only_key",
|
|
tools=[read_doc_id, artifact_id],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
chat_tools = ToolExecutor(user_api_key="wf_only_key", user="alice").get_tools()
|
|
chat_names = {t["name"] for t in chat_tools.values()}
|
|
assert "artifact_generator" in chat_names
|
|
assert "read_document" not in chat_names
|
|
|
|
scoped = ToolExecutor(user="alice")
|
|
scoped.allowed_tool_ids = [read_doc_id]
|
|
node_names = {t["name"] for t in scoped.get_tools().values()}
|
|
assert node_names == {"read_document"}
|
|
|
|
def test_agentless_chat_synthesizes_defaults(self, pg_conn, monkeypatch):
|
|
from docsgpt.agents.default_tools import loaded_default_tools
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
UserToolsRepository(pg_conn).create(
|
|
user_id="alice", name="tool1", status=True
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user="alice")
|
|
tools = executor.get_tools()
|
|
assert isinstance(tools, dict)
|
|
assert len(tools) == 1 + len(loaded_default_tools())
|
|
names = {t["name"] for t in tools.values()}
|
|
assert "tool1" in names
|
|
assert "memory" in names
|
|
|
|
def test_agent_bound_chat_via_user_path_excludes_defaults(
|
|
self, pg_conn, monkeypatch
|
|
):
|
|
"""``agent_id`` forces ``agents.tools``-only; no defaults synthesized."""
|
|
from docsgpt.agents.default_tools import loaded_default_tools
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
UserToolsRepository(pg_conn).create(
|
|
user_id="alice", name="tool1", status=True
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user="alice", agent_id="agent-x")
|
|
tools = executor.get_tools()
|
|
names = {t["name"] for t in tools.values()}
|
|
assert "tool1" in names
|
|
assert not (set(loaded_default_tools()) & names)
|
|
|
|
def test_get_tools_defaults_to_local(self, pg_conn, monkeypatch):
|
|
from docsgpt.agents.default_tools import loaded_default_tools
|
|
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor()
|
|
tools = executor.get_tools()
|
|
assert isinstance(tools, dict)
|
|
assert len(tools) == len(loaded_default_tools())
|
|
assert {t["name"] for t in tools.values()} == set(loaded_default_tools())
|
|
|
|
def test_api_key_path_excludes_defaults(self, pg_conn, monkeypatch):
|
|
"""Agent-bound resolution returns exactly ``agents.tools``."""
|
|
from docsgpt.agents.default_tools import loaded_default_tools
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
tool = UserToolsRepository(pg_conn).create(user_id="alice", name="tool1")
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="alice",
|
|
name="a",
|
|
status="active",
|
|
key="key-agentbound",
|
|
tools=[str(tool["id"])],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user_api_key="key-agentbound", user="alice")
|
|
tools = executor.get_tools()
|
|
names = {t["name"] for t in tools.values()}
|
|
assert names == {"tool1"}
|
|
assert not (set(loaded_default_tools()) & names)
|
|
|
|
def test_api_key_path_empty_agent_tools_gets_nothing(
|
|
self, pg_conn, monkeypatch
|
|
):
|
|
"""Empty ``agents.tools`` invoked via API key yields no tools."""
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="bob",
|
|
name="a",
|
|
status="active",
|
|
key="key-empty",
|
|
tools=[],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user_api_key="key-empty", user="bob")
|
|
assert executor.get_tools() == {}
|
|
|
|
def test_api_key_path_only_synthesizes_author_added_defaults(
|
|
self, pg_conn, monkeypatch
|
|
):
|
|
"""Only ``read_webpage`` in ``agents.tools`` -> exactly that; no other defaults bolted on."""
|
|
from docsgpt.agents.default_tools import default_tool_id
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
|
|
read_webpage_id = default_tool_id("read_webpage")
|
|
memory_id = default_tool_id("memory")
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="erin",
|
|
name="a",
|
|
status="active",
|
|
key="key-only-read",
|
|
tools=[read_webpage_id],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(
|
|
user_api_key="key-only-read", user="erin", agent_id="erin-agent"
|
|
)
|
|
tools = executor.get_tools()
|
|
assert set(tools) == {read_webpage_id}
|
|
assert tools[read_webpage_id]["name"] == "read_webpage"
|
|
assert memory_id not in tools
|
|
assert "memory" not in {t["name"] for t in tools.values()}
|
|
|
|
def test_explicit_default_on_agent_resolves(
|
|
self, pg_conn, monkeypatch
|
|
):
|
|
"""A default tool added explicitly to ``agents.tools`` resolves for every caller."""
|
|
from docsgpt.agents.default_tools import default_tool_id
|
|
from docsgpt.storage.db.repositories.agents import AgentsRepository
|
|
|
|
memory_id = default_tool_id("memory")
|
|
AgentsRepository(pg_conn).create(
|
|
user_id="erin",
|
|
name="a",
|
|
status="active",
|
|
key="key-explicit-default",
|
|
tools=[memory_id],
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(
|
|
user_api_key="key-explicit-default", user="erin"
|
|
)
|
|
tools = executor.get_tools()
|
|
assert set(tools) == {memory_id}
|
|
assert tools[memory_id]["name"] == "memory"
|
|
|
|
def test_no_dedup_between_explicit_and_default_memory(
|
|
self, pg_conn, monkeypatch
|
|
):
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
# Explicit ``memory`` row and the default ``memory`` coexist (separate stores).
|
|
UserToolsRepository(pg_conn).create(
|
|
user_id="dave", name="memory", status=True
|
|
)
|
|
self._patch_conn(monkeypatch, pg_conn)
|
|
|
|
executor = ToolExecutor(user="dave")
|
|
tools = executor.get_tools()
|
|
memory_entries = [t for t in tools.values() if t["name"] == "memory"]
|
|
assert len(memory_entries) == 2
|
|
ids = {t["id"] for t in memory_entries}
|
|
assert len(ids) == 2
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestToolExecutorPrepare:
|
|
|
|
def test_prepare_tools_for_llm_empty(self):
|
|
executor = ToolExecutor()
|
|
result = executor.prepare_tools_for_llm({})
|
|
assert result == []
|
|
|
|
def test_prepare_tools_for_llm_non_api_tool(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "test_tool",
|
|
"actions": [
|
|
{
|
|
"name": "do_thing",
|
|
"description": "Does a thing",
|
|
"active": True,
|
|
"parameters": {
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "The query",
|
|
"filled_by_llm": True,
|
|
"required": True,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
assert len(result) == 1
|
|
assert result[0]["type"] == "function"
|
|
assert result[0]["function"]["name"] == "do_thing"
|
|
assert "query" in result[0]["function"]["parameters"]["properties"]
|
|
|
|
def test_prepare_tools_skips_inactive_actions(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "test_tool",
|
|
"actions": [
|
|
{"name": "active_one", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
{"name": "inactive_one", "description": "D", "active": False, "parameters": {"properties": {}}},
|
|
],
|
|
}
|
|
}
|
|
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
assert len(result) == 1
|
|
assert result[0]["function"]["name"] == "active_one"
|
|
|
|
def test_prepare_tools_builds_name_mapping(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "test_tool",
|
|
"actions": [
|
|
{"name": "do_thing", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
}
|
|
}
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
assert executor._name_to_tool["do_thing"] == ("t1", "do_thing")
|
|
assert executor._tool_to_name[("t1", "do_thing")] == "do_thing"
|
|
|
|
def test_prepare_tools_duplicate_names_get_tool_prefixes(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "tool_a",
|
|
"actions": [
|
|
{"name": "search", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": "tool_b",
|
|
"actions": [
|
|
{"name": "search", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
assert "tool_a_search" in names
|
|
assert "tool_b_search" in names
|
|
assert executor._name_to_tool["tool_a_search"] == ("t1", "search")
|
|
assert executor._name_to_tool["tool_b_search"] == ("t2", "search")
|
|
|
|
def test_prepare_tools_same_named_tools_fall_back_to_numbers(self):
|
|
"""Two tools with the same name (e.g. two MCP rows) still get unique names."""
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "mcp",
|
|
"actions": [
|
|
{"name": "search", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": "mcp",
|
|
"actions": [
|
|
{"name": "search", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
assert "mcp_search" in names
|
|
assert "mcp_search_1" in names
|
|
assert executor._name_to_tool["mcp_search"][1] == "search"
|
|
assert executor._name_to_tool["mcp_search_1"][1] == "search"
|
|
|
|
def test_prepare_tools_unique_name_no_suffix(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "tool_a",
|
|
"actions": [
|
|
{"name": "get_weather", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": "tool_b",
|
|
"actions": [
|
|
{"name": "send_email", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
assert "get_weather" in names
|
|
assert "send_email" in names
|
|
|
|
def test_prepare_tools_prefixed_names_clamped_to_64_chars(self):
|
|
"""Prefixed names must fit the 64-char provider function-name limit."""
|
|
executor = ToolExecutor()
|
|
name_a = "server_" + "x" * 60
|
|
name_b = "server_" + "x" * 60 + "_b" # same first 64 chars as name_a
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": name_a,
|
|
"actions": [
|
|
{"name": "search_documents_in_collection", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": name_b,
|
|
"actions": [
|
|
{"name": "search_documents_in_collection", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
assert all(len(n) <= 64 for n in names)
|
|
assert len(set(names)) == 2
|
|
# Routing still resolves each clamped name to its original action.
|
|
assert {executor._name_to_tool[n] for n in names} == {
|
|
("t1", "search_documents_in_collection"),
|
|
("t2", "search_documents_in_collection"),
|
|
}
|
|
|
|
def test_prepare_tools_long_unique_name_clamped(self):
|
|
"""A unique action name over the limit is truncated, not passed through."""
|
|
executor = ToolExecutor()
|
|
long_action = "fetch_" + "y" * 70
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "tool_a",
|
|
"actions": [
|
|
{"name": long_action, "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
assert names == [long_action[:64]]
|
|
assert executor._name_to_tool[long_action[:64]] == ("t1", long_action)
|
|
|
|
def test_prepare_tools_prefix_skips_collision_with_unique_name(self):
|
|
"""A prefixed candidate must not steal another action's unique name."""
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "tool_a",
|
|
"actions": [
|
|
{"name": "foo", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": "tool_b",
|
|
"actions": [
|
|
{"name": "foo", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
"t3": {
|
|
"name": "tool_c",
|
|
"actions": [
|
|
{"name": "tool_a_foo", "description": "D", "active": True, "parameters": {"properties": {}}},
|
|
],
|
|
},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
names = [r["function"]["name"] for r in result]
|
|
# tool_a_foo is taken by the unique action, so t1's duplicate
|
|
# falls back to a numbered variant of its prefixed name.
|
|
assert "tool_a_foo" in names # The unique action
|
|
assert "tool_a_foo_1" in names
|
|
assert "tool_b_foo" in names
|
|
assert executor._name_to_tool["tool_a_foo"] == ("t3", "tool_a_foo")
|
|
assert executor._name_to_tool["tool_a_foo_1"] == ("t1", "foo")
|
|
|
|
def test_build_tool_parameters_filters_non_llm_fields(self):
|
|
executor = ToolExecutor()
|
|
action = {
|
|
"parameters": {
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "Search query",
|
|
"filled_by_llm": True,
|
|
"value": "default_val",
|
|
"required": True,
|
|
},
|
|
"hidden": {
|
|
"type": "string",
|
|
"filled_by_llm": False,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
result = executor._build_tool_parameters(action)
|
|
assert "query" in result["properties"]
|
|
assert "hidden" not in result["properties"]
|
|
assert "query" in result["required"]
|
|
# filled_by_llm, value, required stripped from schema
|
|
assert "filled_by_llm" not in result["properties"]["query"]
|
|
assert "value" not in result["properties"]["query"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCheckPause:
|
|
|
|
def _make_call(self, name="action_toolid", call_id="c1", arguments="{}"):
|
|
call = Mock()
|
|
call.name = name
|
|
call.id = call_id
|
|
call.arguments = arguments
|
|
call.thought_signature = None
|
|
return call
|
|
|
|
def test_client_side_tool_returns_llm_name(self):
|
|
"""check_pause returns the clean LLM-facing name and llm_name field."""
|
|
executor = ToolExecutor()
|
|
|
|
tools_dict = {
|
|
"ct0": {
|
|
"name": "write_file",
|
|
"client_side": True,
|
|
"actions": [
|
|
{"name": "write_file", "description": "Write a file", "active": True, "parameters": {}},
|
|
],
|
|
}
|
|
}
|
|
|
|
# Prepare tools so the mapping is built
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
|
|
call = self._make_call(name="write_file")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
|
|
assert result is not None
|
|
assert result["name"] == "write_file"
|
|
assert result["llm_name"] == "write_file"
|
|
assert result["action_name"] == "write_file"
|
|
assert result["tool_id"] == "ct0"
|
|
|
|
def test_approval_required_returns_llm_name(self):
|
|
"""check_pause for approval-required tools returns clean LLM name."""
|
|
executor = ToolExecutor()
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "dangerous_tool",
|
|
"actions": [
|
|
{"name": "delete_all", "description": "Deletes everything", "active": True,
|
|
"require_approval": True, "parameters": {}},
|
|
],
|
|
}
|
|
}
|
|
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
|
|
call = self._make_call(name="delete_all")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
|
|
assert result is not None
|
|
assert result["name"] == "delete_all"
|
|
assert result["llm_name"] == "delete_all"
|
|
assert result["action_name"] == "delete_all"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCheckPauseRemoteDevice:
|
|
"""The gate must consult the live ``RemoteDeviceTool`` decision for
|
|
``remote_device`` so ``approval_mode`` changes after pair time and
|
|
per-command heuristics (denylist, sticky) are respected.
|
|
"""
|
|
|
|
def _make_call(self, name="run_command", call_id="c1", command="ls -la /tmp"):
|
|
import json
|
|
|
|
call = Mock()
|
|
call.name = name
|
|
call.id = call_id
|
|
call.arguments = json.dumps({"command": command})
|
|
call.thought_signature = None
|
|
return call
|
|
|
|
def _tools_dict(self, *, device_id="dev_abc"):
|
|
return {
|
|
"rd0": {
|
|
"id": "rd0",
|
|
"name": "remote_device",
|
|
"config": {"device_id": device_id},
|
|
"actions": [
|
|
{
|
|
"name": "run_command",
|
|
"description": "Execute on device",
|
|
"active": True,
|
|
"require_approval": False,
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {
|
|
"type": "string",
|
|
"filled_by_llm": True,
|
|
"value": "",
|
|
},
|
|
},
|
|
"required": ["command"],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
def _patch_device(self, monkeypatch, device):
|
|
"""Stub ``RemoteDeviceTool._load_device`` to return ``device``."""
|
|
from docsgpt.agents.tools import remote_device
|
|
|
|
monkeypatch.setattr(
|
|
remote_device.RemoteDeviceTool,
|
|
"_load_device",
|
|
lambda self: device,
|
|
)
|
|
|
|
def _patch_sticky(self, monkeypatch, patterns):
|
|
"""Stub the sticky lookup to match any normalized pattern in ``patterns``."""
|
|
from docsgpt.agents.tools import remote_device
|
|
|
|
monkeypatch.setattr(
|
|
remote_device,
|
|
"normalize_command",
|
|
lambda cmd: cmd.split()[0] + " *" if cmd else "",
|
|
)
|
|
captured = {"patterns": set(patterns)}
|
|
|
|
class _StubRepo:
|
|
def __init__(self, conn):
|
|
pass
|
|
|
|
def has_pattern(self, device_id, user_id, pattern):
|
|
return pattern in captured["patterns"]
|
|
|
|
monkeypatch.setattr(
|
|
remote_device,
|
|
"DeviceAutoApprovePatternsRepository",
|
|
_StubRepo,
|
|
)
|
|
|
|
class _StubConn:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
monkeypatch.setattr(remote_device, "db_readonly", lambda: _StubConn())
|
|
|
|
def test_full_non_deny_no_pause(self, monkeypatch):
|
|
"""``full`` + a non-denylisted command auto-approves."""
|
|
executor = ToolExecutor(user="alice")
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{
|
|
"id": "dev_abc",
|
|
"approval_mode": "full",
|
|
"status": "active",
|
|
},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="whoami")
|
|
assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
|
|
|
|
def test_full_writing_command_no_pause(self, monkeypatch):
|
|
"""``full`` auto-approves writes too (only the denylist stops it)."""
|
|
executor = ToolExecutor(user="alice")
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{
|
|
"id": "dev_abc",
|
|
"approval_mode": "full",
|
|
"status": "active",
|
|
},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="rm /tmp/foo")
|
|
assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
|
|
|
|
def test_full_denylist_forces_pause(self, monkeypatch):
|
|
"""``full`` + a denylisted command still pauses (forced prompt)."""
|
|
executor = ToolExecutor(user="alice")
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{
|
|
"id": "dev_abc",
|
|
"approval_mode": "full",
|
|
"status": "active",
|
|
},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="rm -rf /")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
assert result is not None
|
|
assert result["pause_type"] == "awaiting_approval"
|
|
|
|
def test_ask_mode_pauses(self, monkeypatch):
|
|
"""``ask`` + any command pauses by default."""
|
|
executor = ToolExecutor(user="alice")
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{
|
|
"id": "dev_abc",
|
|
"approval_mode": "ask",
|
|
"status": "active",
|
|
},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="ls -la /tmp")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
assert result is not None
|
|
assert result["pause_type"] == "awaiting_approval"
|
|
|
|
def test_ask_mode_sticky_match_no_pause(self, monkeypatch):
|
|
"""``ask`` + a command matching a stored sticky pattern auto-approves."""
|
|
executor = ToolExecutor(user="alice")
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{
|
|
"id": "dev_abc",
|
|
"approval_mode": "ask",
|
|
"status": "active",
|
|
},
|
|
)
|
|
# The stub normalizer turns ``ls -la /tmp`` into ``ls *``.
|
|
self._patch_sticky(monkeypatch, {"ls *"})
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="ls -la /tmp")
|
|
assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCheckPauseRemoteDeviceHeadless:
|
|
"""Headless gating for ``remote_device``.
|
|
|
|
A denylist-forced prompt must NOT be bypassed by the run's
|
|
``tool_allowlist`` — otherwise a scheduled/headless run with the device
|
|
allowlisted would auto-execute even a denylisted command. Normal
|
|
(non-forced) approvals keep the allowlist bypass.
|
|
"""
|
|
|
|
def _make_call(self, name="run_command", call_id="c1", command="whoami"):
|
|
import json
|
|
|
|
call = Mock()
|
|
call.name = name
|
|
call.id = call_id
|
|
call.arguments = json.dumps({"command": command})
|
|
call.thought_signature = None
|
|
return call
|
|
|
|
def _tools_dict(self, *, device_id="dev_abc"):
|
|
return {
|
|
"rd0": {
|
|
"id": "rd0",
|
|
"name": "remote_device",
|
|
"config": {"device_id": device_id},
|
|
"actions": [
|
|
{
|
|
"name": "run_command",
|
|
"description": "Execute on device",
|
|
"active": True,
|
|
"require_approval": False,
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {
|
|
"type": "string",
|
|
"filled_by_llm": True,
|
|
"value": "",
|
|
},
|
|
},
|
|
"required": ["command"],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
def _patch_device(self, monkeypatch, device):
|
|
from docsgpt.agents.tools import remote_device
|
|
|
|
monkeypatch.setattr(
|
|
remote_device.RemoteDeviceTool,
|
|
"_load_device",
|
|
lambda self: device,
|
|
)
|
|
|
|
def _patch_sticky(self, monkeypatch, patterns):
|
|
from docsgpt.agents.tools import remote_device
|
|
|
|
monkeypatch.setattr(
|
|
remote_device,
|
|
"normalize_command",
|
|
lambda cmd: cmd.split()[0] + " *" if cmd else "",
|
|
)
|
|
captured = {"patterns": set(patterns)}
|
|
|
|
class _StubRepo:
|
|
def __init__(self, conn):
|
|
pass
|
|
|
|
def has_pattern(self, device_id, user_id, pattern):
|
|
return pattern in captured["patterns"]
|
|
|
|
monkeypatch.setattr(
|
|
remote_device, "DeviceAutoApprovePatternsRepository", _StubRepo
|
|
)
|
|
|
|
class _StubConn:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
monkeypatch.setattr(remote_device, "db_readonly", lambda: _StubConn())
|
|
|
|
def test_headless_allowlisted_denylisted_command_denied(self, monkeypatch):
|
|
"""Device allowlisted, but ``rm -rf /`` is a forced prompt -> denied."""
|
|
executor = ToolExecutor(
|
|
user="alice", headless=True, tool_allowlist={"rd0"}
|
|
)
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{"id": "dev_abc", "approval_mode": "full", "status": "active"},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="rm -rf /")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
assert result is not None
|
|
assert result["pause_type"] == "headless_denied"
|
|
assert result["error_type"] == "tool_not_allowed"
|
|
|
|
def test_headless_allowlisted_normal_command_executes(self, monkeypatch):
|
|
"""Device allowlisted + a normal command in ``full`` -> executes."""
|
|
executor = ToolExecutor(
|
|
user="alice", headless=True, tool_allowlist={"rd0"}
|
|
)
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{"id": "dev_abc", "approval_mode": "full", "status": "active"},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="whoami")
|
|
assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
|
|
|
|
def test_headless_not_allowlisted_denied(self, monkeypatch):
|
|
"""Without the allowlist, an ask-mode command is denied headless."""
|
|
executor = ToolExecutor(user="alice", headless=True, tool_allowlist=set())
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{"id": "dev_abc", "approval_mode": "ask", "status": "active"},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="ls -la /tmp")
|
|
result = executor.check_pause(tools_dict, call, "OpenAILLM")
|
|
assert result is not None
|
|
assert result["pause_type"] == "headless_denied"
|
|
|
|
def test_headless_allowlisted_ask_mode_normal_executes(self, monkeypatch):
|
|
"""ask-mode + allowlisted + non-denylisted -> bypass allowed."""
|
|
executor = ToolExecutor(
|
|
user="alice", headless=True, tool_allowlist={"rd0"}
|
|
)
|
|
self._patch_device(
|
|
monkeypatch,
|
|
{"id": "dev_abc", "approval_mode": "ask", "status": "active"},
|
|
)
|
|
self._patch_sticky(monkeypatch, set())
|
|
|
|
tools_dict = self._tools_dict()
|
|
executor.prepare_tools_for_llm(tools_dict)
|
|
call = self._make_call(command="ls -la /tmp")
|
|
# Not denylist-forced, so the allowlist bypass applies.
|
|
assert executor.check_pause(tools_dict, call, "OpenAILLM") is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestToolExecutorExecute:
|
|
|
|
def _make_call(self, name="action_toolid", call_id="c1", arguments="{}"):
|
|
call = Mock()
|
|
call.name = name
|
|
call.id = call_id
|
|
call.arguments = arguments
|
|
return call
|
|
|
|
def test_execute_parse_failure(self, monkeypatch):
|
|
executor = ToolExecutor()
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(parse_args=Mock(return_value=(None, None, {}))),
|
|
)
|
|
|
|
call = self._make_call(name="bad")
|
|
gen = executor.execute({}, call, "MockLLM")
|
|
|
|
events = []
|
|
result = None
|
|
while True:
|
|
try:
|
|
events.append(next(gen))
|
|
except StopIteration as e:
|
|
result = e.value
|
|
break
|
|
|
|
# The bare "Failed to parse tool call." gave the model nothing to act
|
|
# on, so it retried the same invented call until the iteration cap.
|
|
assert "Could not run 'bad'" in result[0]
|
|
assert "Available tools:" in result[0]
|
|
assert len(executor.tool_calls) == 1
|
|
assert events[0]["data"]["status"] == "error"
|
|
|
|
def test_execute_tool_not_found(self, monkeypatch):
|
|
executor = ToolExecutor()
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(parse_args=Mock(return_value=("missing_id", "action", {}))),
|
|
)
|
|
|
|
call = self._make_call()
|
|
gen = executor.execute({}, call, "MockLLM")
|
|
|
|
events = []
|
|
result = None
|
|
while True:
|
|
try:
|
|
events.append(next(gen))
|
|
except StopIteration as e:
|
|
result = e.value
|
|
break
|
|
|
|
# The message must name the offending call and what can be called
|
|
# instead, so the model has something to correct against.
|
|
assert "no such tool" in result[0]
|
|
assert "Available tools:" in result[0]
|
|
assert events[0]["data"]["status"] == "error"
|
|
|
|
def test_execute_success(self, mock_tool_manager, monkeypatch):
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(parse_args=Mock(return_value=("t1", "test_action", {"param1": "val"}))),
|
|
)
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "test_tool",
|
|
"config": {"key": "val"},
|
|
"actions": [
|
|
{"name": "test_action", "description": "Test", "parameters": {"properties": {}}},
|
|
],
|
|
}
|
|
}
|
|
|
|
call = self._make_call(name="test_action_t1", call_id="c1")
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
|
|
events = []
|
|
result = None
|
|
while True:
|
|
try:
|
|
events.append(next(gen))
|
|
except StopIteration as e:
|
|
result = e.value
|
|
break
|
|
|
|
assert result[0] == "Tool result"
|
|
assert result[1] == "c1"
|
|
|
|
statuses = [e["data"]["status"] for e in events]
|
|
assert "pending" in statuses
|
|
assert "completed" in statuses
|
|
|
|
def test_get_truncated_tool_calls(self):
|
|
from docsgpt.agents.tool_executor import PERSISTED_RESULT_MAX_LEN
|
|
|
|
executor = ToolExecutor()
|
|
executor.tool_calls = [
|
|
{
|
|
"tool_name": "test",
|
|
"call_id": "1",
|
|
"action_name": "act",
|
|
"arguments": {},
|
|
"result": "A" * 100,
|
|
},
|
|
{
|
|
"tool_name": "test",
|
|
"call_id": "2",
|
|
"action_name": "act",
|
|
"arguments": {},
|
|
"result": "B" * (PERSISTED_RESULT_MAX_LEN + 100),
|
|
},
|
|
]
|
|
|
|
truncated = executor.get_truncated_tool_calls()
|
|
assert len(truncated) == 2
|
|
# Short results persist verbatim so stored errors stay diagnosable.
|
|
assert truncated[0]["result"] == "A" * 100
|
|
assert truncated[0]["status"] == "completed"
|
|
# Oversize results are bounded for the message JSONB copy.
|
|
assert len(truncated[1]["result"]) == PERSISTED_RESULT_MAX_LEN + 3
|
|
assert truncated[1]["result"].endswith("...")
|
|
|
|
def test_get_truncated_tool_calls_preserves_error_status(self):
|
|
executor = ToolExecutor()
|
|
executor.tool_calls = [
|
|
{
|
|
"tool_name": "test",
|
|
"call_id": "1",
|
|
"action_name": "act",
|
|
"arguments": {},
|
|
"result": "boom",
|
|
"status": "error",
|
|
}
|
|
]
|
|
|
|
truncated = executor.get_truncated_tool_calls()
|
|
assert truncated[0]["status"] == "error"
|
|
|
|
def test_result_status_reflects_in_band_tool_errors(self):
|
|
from docsgpt.agents.tool_executor import result_status
|
|
|
|
assert result_status({"status": "error", "error": "invalid spec"}) == "error"
|
|
assert result_status({"error": "input artifact A9 not found"}) == "error"
|
|
assert result_status({"status": "ok", "artifact_id": "x"}) == "completed"
|
|
assert result_status({"error": None, "status": "ok"}) == "completed"
|
|
assert result_status("plain text result") == "completed"
|
|
|
|
def test_tool_caching(self, mock_tool_manager, monkeypatch):
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(parse_args=Mock(return_value=("t1", "test_action", {}))),
|
|
)
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "test_tool",
|
|
"config": {"key": "val"},
|
|
"actions": [
|
|
{"name": "test_action", "description": "Test", "parameters": {"properties": {}}},
|
|
],
|
|
}
|
|
}
|
|
|
|
call = self._make_call(name="test_action_t1")
|
|
|
|
# First execution — loads tool
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
while True:
|
|
try:
|
|
next(gen)
|
|
except StopIteration:
|
|
break
|
|
|
|
# Second execution — should use cache
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
while True:
|
|
try:
|
|
next(gen)
|
|
except StopIteration:
|
|
break
|
|
|
|
# load_tool called only once due to cache
|
|
assert mock_tool_manager.load_tool.call_count == 1
|
|
|
|
def test_execute_api_tool(self, mock_tool_manager, monkeypatch):
|
|
"""Cover lines 199-202, 256-267: api_tool execution path."""
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(
|
|
parse_args=Mock(return_value=("t1", "get_users", {"body_param": "val"}))
|
|
),
|
|
)
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "api_tool",
|
|
"config": {
|
|
"actions": {
|
|
"get_users": {
|
|
"name": "get_users",
|
|
"description": "Get users",
|
|
"url": "https://api.example.com/users",
|
|
"method": "GET",
|
|
"query_params": {"properties": {}},
|
|
"headers": {"properties": {}},
|
|
"body": {"properties": {}},
|
|
"active": True,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
}
|
|
|
|
call = self._make_call(name="get_users_t1", call_id="c2")
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
|
|
events = []
|
|
result = None
|
|
while True:
|
|
try:
|
|
events.append(next(gen))
|
|
except StopIteration as e:
|
|
result = e.value
|
|
break
|
|
|
|
assert result is not None
|
|
statuses = [e["data"]["status"] for e in events]
|
|
assert "pending" in statuses
|
|
|
|
def test_execute_with_prefilled_param_values(self, mock_tool_manager, monkeypatch):
|
|
"""Cover line 179: params not in call_args use default value."""
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(
|
|
parse_args=Mock(return_value=("t1", "act", {}))
|
|
),
|
|
)
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "test_tool",
|
|
"config": {"key": "val"},
|
|
"actions": [
|
|
{
|
|
"name": "act",
|
|
"description": "Test",
|
|
"parameters": {
|
|
"properties": {
|
|
"hidden_param": {
|
|
"type": "string",
|
|
"value": "default_val",
|
|
"filled_by_llm": False,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
call = self._make_call(name="act_t1")
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
|
|
while True:
|
|
try:
|
|
next(gen)
|
|
except StopIteration as e:
|
|
result = e.value
|
|
break
|
|
|
|
assert result[0] == "Tool result"
|
|
|
|
def test_execute_tool_with_artifact_id(self, mock_tool_manager, monkeypatch):
|
|
"""Cover lines 217-218: tool with get_artifact_id."""
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolActionParser",
|
|
lambda _cls, **kw: Mock(
|
|
parse_args=Mock(return_value=("t1", "act", {"q": "v"}))
|
|
),
|
|
)
|
|
|
|
mock_tool = mock_tool_manager.load_tool.return_value
|
|
mock_tool.get_artifact_id = Mock(return_value="artifact-123")
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "test_tool",
|
|
"config": {"key": "val"},
|
|
"actions": [
|
|
{
|
|
"name": "act",
|
|
"description": "Test",
|
|
"parameters": {"properties": {}},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
|
|
call = self._make_call(name="act_t1")
|
|
gen = executor.execute(tools_dict, call, "MockLLM")
|
|
|
|
events = []
|
|
while True:
|
|
try:
|
|
events.append(next(gen))
|
|
except StopIteration:
|
|
break
|
|
|
|
completed_events = [
|
|
e for e in events if e["data"].get("status") == "completed"
|
|
]
|
|
assert any(
|
|
"artifact_id" in e.get("data", {}) for e in completed_events
|
|
)
|
|
|
|
def test_get_or_load_tool_encrypted_credentials(self, monkeypatch):
|
|
"""Cover lines 273-278: encrypted credentials path."""
|
|
executor = ToolExecutor(user="test_user")
|
|
|
|
mock_tm = Mock()
|
|
mock_tool = Mock()
|
|
mock_tm.load_tool.return_value = mock_tool
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolManager", lambda config: mock_tm
|
|
)
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.decrypt_credentials",
|
|
lambda creds, user: {"api_key": "decrypted_key"},
|
|
)
|
|
|
|
tool_data = {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "custom_tool",
|
|
"config": {"encrypted_credentials": "encrypted_blob"},
|
|
}
|
|
|
|
result = executor._get_or_load_tool(tool_data, "t1", "act")
|
|
assert result is mock_tool
|
|
call_kwargs = mock_tm.load_tool.call_args
|
|
tool_config = call_kwargs[1]["tool_config"] if "tool_config" in call_kwargs[1] else call_kwargs[0][1]
|
|
assert "api_key" in tool_config.get("auth_credentials", tool_config)
|
|
|
|
def test_get_or_load_tool_decrypts_with_tool_owner(self, monkeypatch):
|
|
"""Team-shared tool: credentials decrypt with the OWNER's sub, not the
|
|
invoker's (teams OQ2 — delegation). The tool row carries user_id=owner
|
|
while the executor runs as a different member."""
|
|
executor = ToolExecutor(user="member_bob")
|
|
|
|
mock_tm = Mock()
|
|
mock_tm.load_tool.return_value = Mock()
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolManager", lambda config: mock_tm
|
|
)
|
|
captured = {}
|
|
|
|
def _fake_decrypt(creds, user):
|
|
captured["user"] = user
|
|
return {"api_key": "owner_secret"}
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.decrypt_credentials", _fake_decrypt
|
|
)
|
|
|
|
tool_data = {
|
|
"id": "00000000-0000-0000-0000-000000000009",
|
|
"name": "custom_tool",
|
|
"user_id": "owner_alice",
|
|
"config": {"encrypted_credentials": "blob"},
|
|
}
|
|
executor._get_or_load_tool(tool_data, "t9", "act")
|
|
# Decrypted under the tool owner, NOT the invoking member.
|
|
assert captured["user"] == "owner_alice"
|
|
|
|
def test_get_or_load_tool_mcp_tool(self, monkeypatch):
|
|
"""Cover lines 281-283: mcp_tool path sets query_mode."""
|
|
executor = ToolExecutor(user="test_user")
|
|
executor.conversation_id = "conv-123"
|
|
|
|
mock_tm = Mock()
|
|
mock_tool = Mock()
|
|
mock_tm.load_tool.return_value = mock_tool
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolManager", lambda config: mock_tm
|
|
)
|
|
|
|
tool_data = {
|
|
"id": "00000000-0000-0000-0000-000000000002",
|
|
"name": "mcp_tool",
|
|
"config": {},
|
|
}
|
|
|
|
result = executor._get_or_load_tool(tool_data, "t1", "act")
|
|
assert result is mock_tool
|
|
call_kwargs = mock_tm.load_tool.call_args
|
|
tool_config = call_kwargs[1].get("tool_config", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else {})
|
|
assert tool_config.get("query_mode") is True
|
|
assert tool_config.get("conversation_id") == "conv-123"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coverage — additional uncovered lines: 217-218, 256-267
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestToolExecutorAdditionalCoverage:
|
|
|
|
def test_get_artifact_id_exception_handled(self, monkeypatch):
|
|
"""Cover lines 217-218: get_artifact_id raises exception."""
|
|
from types import SimpleNamespace
|
|
|
|
executor = ToolExecutor(user="user1")
|
|
|
|
mock_tool = Mock()
|
|
mock_tool.execute_action.return_value = "result"
|
|
mock_tool.get_artifact_id.side_effect = RuntimeError("artifact error")
|
|
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolManager",
|
|
lambda config: Mock(load_tool=Mock(return_value=mock_tool)),
|
|
)
|
|
|
|
tools_dict = {
|
|
"t1": {
|
|
"id": "00000000-0000-0000-0000-000000000001",
|
|
"name": "custom_tool",
|
|
"config": {"key": "val"},
|
|
"actions": [
|
|
{
|
|
"name": "action1",
|
|
"active": True,
|
|
"parameters": {"properties": {}},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
# Create a fake call object matching what ToolActionParser expects
|
|
call = SimpleNamespace(
|
|
id="c1",
|
|
function=SimpleNamespace(
|
|
name="action1_t1",
|
|
arguments="{}",
|
|
),
|
|
)
|
|
events = list(executor.execute(tools_dict, call, "OpenAILLM"))
|
|
# Should complete without raising; artifact_id error is logged but not raised
|
|
assert any(
|
|
isinstance(e, dict) and e.get("type") == "tool_call"
|
|
for e in events
|
|
)
|
|
|
|
def test_get_or_load_api_tool_with_body_content_type(self, monkeypatch):
|
|
"""Cover lines 256-267: api_tool with body_content_type."""
|
|
executor = ToolExecutor(user="user1")
|
|
|
|
mock_tm = Mock()
|
|
mock_tool = Mock()
|
|
mock_tm.load_tool.return_value = mock_tool
|
|
monkeypatch.setattr(
|
|
"docsgpt.agents.tool_executor.ToolManager", lambda config: mock_tm
|
|
)
|
|
|
|
tool_data = {
|
|
"name": "api_tool",
|
|
"config": {
|
|
"actions": {
|
|
"create": {
|
|
"url": "https://api.example.com/items",
|
|
"method": "POST",
|
|
"body_content_type": "application/json",
|
|
"body_encoding_rules": {"encode_as": "json"},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
result = executor._get_or_load_tool(
|
|
tool_data, "t1", "create",
|
|
headers={"Authorization": "Bearer tok"},
|
|
query_params={"page": "1"},
|
|
)
|
|
assert result is mock_tool
|
|
# Verify config was built with body_content_type
|
|
call_args = mock_tm.load_tool.call_args
|
|
tool_config = call_args[1].get("tool_config", call_args[0][1] if len(call_args[0]) > 1 else {})
|
|
assert tool_config.get("body_content_type") == "application/json"
|
|
assert tool_config.get("body_encoding_rules") == {"encode_as": "json"}
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetEnabledToolNames:
|
|
def test_returns_names_from_get_tools(self, monkeypatch):
|
|
executor = ToolExecutor()
|
|
monkeypatch.setattr(
|
|
executor,
|
|
"get_tools",
|
|
lambda: {
|
|
"id1": {"name": "code_executor"},
|
|
"id2": {"name": "search"},
|
|
"id3": {}, # nameless rows are skipped
|
|
},
|
|
)
|
|
assert executor.get_enabled_tool_names() == {"code_executor", "search"}
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDuplicateRegistrationCollapses:
|
|
"""The same builtin registered twice must not reach the model twice.
|
|
|
|
``code_executor`` and ``artifact_generator`` are dual-registered (default
|
|
chat tool *and* builtin agent tool). When both rows resolve, the model is
|
|
handed two indistinguishable copies of every action and their names get
|
|
mangled — one production user's whole session ran on
|
|
``artifact_generator_create_artifact_1`` and ``code_executor_run_code_1``.
|
|
"""
|
|
|
|
ACTION = {
|
|
"name": "create_artifact",
|
|
"description": "Render a document.",
|
|
"active": True,
|
|
"parameters": {"properties": {}},
|
|
}
|
|
|
|
def test_identical_tool_registered_twice_yields_one_clean_name(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"id-default": {"name": "artifact_generator", "actions": [dict(self.ACTION)]},
|
|
"id-builtin": {"name": "artifact_generator", "actions": [dict(self.ACTION)]},
|
|
}
|
|
result = executor.prepare_tools_for_llm(tools_dict)
|
|
|
|
names = [entry["function"]["name"] for entry in result]
|
|
assert names == ["create_artifact"], names
|
|
assert executor._name_to_tool["create_artifact"][1] == "create_artifact"
|
|
|
|
def test_same_action_name_on_different_tools_still_disambiguates(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"t1": {
|
|
"name": "brave",
|
|
"actions": [
|
|
{"name": "search", "description": "Brave.", "active": True, "parameters": {"properties": {}}}
|
|
],
|
|
},
|
|
"t2": {
|
|
"name": "duckduckgo",
|
|
"actions": [
|
|
{"name": "search", "description": "DDG.", "active": True, "parameters": {"properties": {}}}
|
|
],
|
|
},
|
|
}
|
|
names = [entry["function"]["name"] for entry in executor.prepare_tools_for_llm(tools_dict)]
|
|
assert sorted(names) == ["brave_search", "duckduckgo_search"]
|
|
|
|
def test_a_builtin_collapses_even_when_the_metadata_differs(self):
|
|
"""The two rows are never byte-identical, so the rule is name-based.
|
|
|
|
A stored row's actions have been through ``transform_actions`` and this
|
|
change set also edits the tool descriptions, so any stored row created
|
|
earlier has drifted from the synthesized one permanently.
|
|
"""
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"0": {
|
|
"name": "artifact_generator",
|
|
"actions": [
|
|
{"name": "create_artifact", "description": "Old text.", "active": True,
|
|
"parameters": {"properties": {}}}
|
|
],
|
|
},
|
|
"id-synth": {
|
|
"name": "artifact_generator",
|
|
"actions": [
|
|
{"name": "create_artifact", "description": "New text.", "active": True,
|
|
"parameters": {"properties": {}}}
|
|
],
|
|
},
|
|
}
|
|
names = [r["function"]["name"] for r in executor.prepare_tools_for_llm(tools_dict)]
|
|
assert names == ["create_artifact"], names
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestStoredAndSynthesizedBuiltinCollapse:
|
|
"""The real duplicate is a stored row plus the synthesized default.
|
|
|
|
Both synthesizers mint the same uuid5, so two *synthesized* rows can never
|
|
coexist in a dict keyed by id. The duplicate that reached production is the
|
|
user's stored row (keyed by list index) alongside the synthesized default —
|
|
and a stored row has been through ``transform_actions``, so the two are not
|
|
byte-identical. A fingerprint over the action metadata therefore never
|
|
matched and the collapse was a no-op for the only case it was written for.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _stored_action():
|
|
# What transform_actions produces for a stored row.
|
|
return {
|
|
"name": "create_artifact",
|
|
"description": "Render a document.",
|
|
"active": True,
|
|
"parameters": {
|
|
"properties": {"kind": {"type": "string", "filled_by_llm": True, "value": ""}}
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def _synthesized_action():
|
|
return {
|
|
"name": "create_artifact",
|
|
"description": "Render a document.",
|
|
"active": True,
|
|
"parameters": {"properties": {"kind": {"type": "string"}}},
|
|
}
|
|
|
|
def test_stored_row_and_synthesized_default_collapse_to_one(self):
|
|
executor = ToolExecutor()
|
|
tools_dict = {
|
|
"0": {"name": "artifact_generator", "actions": [self._stored_action()]},
|
|
"981e1888-a32f-587b-a306-fba4f6f83e67": {
|
|
"name": "artifact_generator",
|
|
"actions": [self._synthesized_action()],
|
|
},
|
|
}
|
|
names = [r["function"]["name"] for r in executor.prepare_tools_for_llm(tools_dict)]
|
|
assert names == ["create_artifact"], names
|
|
# The stored row wins: it carries the real id and any config.
|
|
assert executor._name_to_tool["create_artifact"] == ("0", "create_artifact")
|
|
|
|
def test_non_builtin_duplicates_are_never_collapsed(self):
|
|
"""Two MCP rows may legitimately share a name, identical metadata or not."""
|
|
executor = ToolExecutor()
|
|
action = {"name": "search", "description": "D", "active": True, "parameters": {"properties": {}}}
|
|
tools_dict = {
|
|
"t1": {"name": "mcp_tool", "actions": [dict(action)]},
|
|
"t2": {"name": "mcp_tool", "actions": [dict(action)]},
|
|
}
|
|
assert len(executor.prepare_tools_for_llm(tools_dict)) == 2
|