test_skill_toolset.py

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import collections
import logging
import sys
from unittest import mock

from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.code_executors.base_code_executor import BaseCodeExecutor
from google.adk.code_executors.code_execution_utils import CodeExecutionResult
from google.adk.code_executors.unsafe_local_code_executor import UnsafeLocalCodeExecutor
from google.adk.models import llm_request as llm_request_model
from google.adk.skills import models
from google.adk.tools import skill_toolset
from google.adk.tools import tool_context
from google.genai import types
import pytest


@pytest.fixture(name="mock_skill1_frontmatter")
def _mock_skill1_frontmatter():
  """Fixture for skill1 frontmatter."""
  frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter.name = "skill1"
  frontmatter.description = "Skill 1 description"
  frontmatter.allowed_tools = ["test_tool"]
  frontmatter.model_dump.return_value = {
      "name": "skill1",
      "description": "Skill 1 description",
  }
  return frontmatter


@pytest.fixture(name="mock_skill1")
def _mock_skill1(mock_skill1_frontmatter):
  """Fixture for skill1."""
  skill = mock.create_autospec(models.Skill, instance=True)
  skill.name = "skill1"
  skill.description = "Skill 1 description"
  skill.instructions = "instructions for skill1"
  skill.frontmatter = mock_skill1_frontmatter
  skill.resources = mock.MagicMock(
      spec=[
          "get_reference",
          "get_asset",
          "get_script",
          "list_references",
          "list_assets",
          "list_scripts",
      ]
  )

  def get_ref(name):
    if name == "ref1.md":
      return "ref content 1"
    if name == "doc.pdf":
      return b"fake pdf content"
    return None

  def get_asset(name):
    if name == "asset1.txt":
      return "asset content 1"
    if name == "image.png":
      return b"fake image content"
    return None

  def get_script(name):
    if name == "setup.sh":
      return models.Script(src="echo setup")
    if name == "run.py":
      return models.Script(src="print('hello')")
    if name == "build.rb":
      return models.Script(src="puts 'hello'")
    return None

  skill.resources.get_reference.side_effect = get_ref
  skill.resources.get_asset.side_effect = get_asset
  skill.resources.get_script.side_effect = get_script
  skill.resources.list_references.return_value = ["ref1.md", "doc.pdf"]
  skill.resources.list_assets.return_value = ["asset1.txt", "image.png"]
  skill.resources.list_scripts.return_value = [
      "setup.sh",
      "run.py",
      "build.rb",
  ]
  return skill


@pytest.fixture(name="mock_skill2_frontmatter")
def _mock_skill2_frontmatter():
  """Fixture for skill2 frontmatter."""
  frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter.name = "skill2"
  frontmatter.description = "Skill 2 description"
  frontmatter.allowed_tools = []
  frontmatter.model_dump.return_value = {
      "name": "skill2",
      "description": "Skill 2 description",
  }
  return frontmatter


@pytest.fixture(name="mock_skill2")
def _mock_skill2(mock_skill2_frontmatter):
  """Fixture for skill2."""
  skill = mock.create_autospec(models.Skill, instance=True)
  skill.name = "skill2"
  skill.description = "Skill 2 description"
  skill.instructions = "instructions for skill2"
  skill.frontmatter = mock_skill2_frontmatter
  skill.resources = mock.MagicMock(
      spec=[
          "get_reference",
          "get_asset",
          "get_script",
          "list_references",
          "list_assets",
          "list_scripts",
      ]
  )

  def get_ref(name):
    if name == "ref2.md":
      return "ref content 2"
    return None

  def get_asset(name):
    if name == "asset2.txt":
      return "asset content 2"
    return None

  skill.resources.get_reference.side_effect = get_ref
  skill.resources.get_asset.side_effect = get_asset
  skill.resources.list_references.return_value = ["ref2.md"]
  skill.resources.list_assets.return_value = ["asset2.txt"]
  skill.resources.list_scripts.return_value = []
  return skill


@pytest.fixture
def tool_context_instance():
  """Fixture for tool context."""
  ctx = mock.create_autospec(tool_context.ToolContext, instance=True)
  ctx._invocation_context = mock.MagicMock()
  ctx._invocation_context.agent = mock.MagicMock()
  ctx._invocation_context.agent.name = "test_agent"
  ctx._invocation_context.agent_states = {}
  ctx.agent_name = "test_agent"
  return ctx


# SkillToolset tests
def test_get_skill(mock_skill1, mock_skill2):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  assert toolset._get_skill("skill1") == mock_skill1
  assert toolset._get_skill("nonexistent") is None


def test_list_skills(mock_skill1, mock_skill2):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  skills = toolset._list_skills()
  assert len(skills) == 2
  assert mock_skill1 in skills
  assert mock_skill2 in skills


@pytest.mark.asyncio
async def test_get_tools(mock_skill1, mock_skill2):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  tools = await toolset.get_tools()
  assert len(tools) == 4
  assert isinstance(tools[0], skill_toolset.ListSkillsTool)
  assert isinstance(tools[1], skill_toolset.LoadSkillTool)
  assert isinstance(tools[2], skill_toolset.LoadSkillResourceTool)
  assert isinstance(tools[3], skill_toolset.RunSkillScriptTool)


@pytest.mark.asyncio
async def test_resolve_additional_tools_from_state_none(mock_skill1):
  toolset = skill_toolset.SkillToolset([mock_skill1])

  # Mock ReadonlyContext
  readonly_context = mock.create_autospec(ReadonlyContext, instance=True)
  readonly_context.agent_name = "test_agent"
  readonly_context.state.get.return_value = None

  result = await toolset._resolve_additional_tools_from_state(readonly_context)

  assert not result


@pytest.mark.asyncio
async def test_list_skills_tool(
    mock_skill1, mock_skill2, tool_context_instance
):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  tool = skill_toolset.ListSkillsTool(toolset)
  result = await tool.run_async(args={}, tool_context=tool_context_instance)
  assert "<available_skills>" in result
  assert "skill1" in result
  assert "skill2" in result


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "args, expected_result",
    [
        (
            {"skill_name": "skill1"},
            {
                "skill_name": "skill1",
                "instructions": "instructions for skill1",
                "frontmatter": {
                    "name": "skill1",
                    "description": "Skill 1 description",
                },
            },
        ),
        (
            {"skill_name": "nonexistent"},
            {
                "error": "Skill 'nonexistent' not found.",
                "error_code": "SKILL_NOT_FOUND",
            },
        ),
        (
            {},
            {
                "error": "Argument 'skill_name' is required.",
                "error_code": "INVALID_ARGUMENTS",
            },
        ),
    ],
)
async def test_load_skill_run_async(
    mock_skill1, tool_context_instance, args, expected_result
):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.LoadSkillTool(toolset)
  result = await tool.run_async(args=args, tool_context=tool_context_instance)
  assert result == expected_result


@pytest.mark.asyncio
async def test_load_skill_run_async_state_none(
    mock_skill1, tool_context_instance
):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.LoadSkillTool(toolset)

  # Mock state to return None for the key
  state_key = "_adk_activated_skill_test_agent"
  tool_context_instance.state.get.return_value = None

  result = await tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )

  assert result["skill_name"] == "skill1"
  tool_context_instance.state.__setitem__.assert_called_with(
      state_key, ["skill1"]
  )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "args, expected_result",
    [
        (
            {"skill_name": "skill1", "file_path": "references/ref1.md"},
            {
                "skill_name": "skill1",
                "file_path": "references/ref1.md",
                "content": "ref content 1",
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "assets/asset1.txt"},
            {
                "skill_name": "skill1",
                "file_path": "assets/asset1.txt",
                "content": "asset content 1",
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "references/doc.pdf"},
            {
                "skill_name": "skill1",
                "file_path": "references/doc.pdf",
                "status": (
                    "Binary file detected. The content has been injected into"
                    " the conversation history for you to analyze."
                ),
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "assets/image.png"},
            {
                "skill_name": "skill1",
                "file_path": "assets/image.png",
                "status": (
                    "Binary file detected. The content has been injected into"
                    " the conversation history for you to analyze."
                ),
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "scripts/setup.sh"},
            {
                "skill_name": "skill1",
                "file_path": "scripts/setup.sh",
                "content": "echo setup",
            },
        ),
        (
            {"skill_name": "nonexistent", "file_path": "references/ref1.md"},
            {
                "error": "Skill 'nonexistent' not found.",
                "error_code": "SKILL_NOT_FOUND",
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "references/other.md"},
            {
                "error": (
                    "Resource 'references/other.md' not found in skill"
                    " 'skill1'."
                ),
                "error_code": "RESOURCE_NOT_FOUND",
            },
        ),
        (
            {"skill_name": "skill1", "file_path": "invalid/path.txt"},
            {
                "error": (
                    "Path must start with 'references/', 'assets/',"
                    " or 'scripts/'."
                ),
                "error_code": "INVALID_RESOURCE_PATH",
            },
        ),
        (
            {"file_path": "references/ref1.md"},
            {
                "error": "Argument 'skill_name' is required.",
                "error_code": "INVALID_ARGUMENTS",
            },
        ),
        (
            {"skill_name": "skill1"},
            {
                "error": "Argument 'file_path' is required.",
                "error_code": "INVALID_ARGUMENTS",
            },
        ),
    ],
)
async def test_load_resource_run_async(
    mock_skill1, tool_context_instance, args, expected_result
):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.LoadSkillResourceTool(toolset)
  result = await tool.run_async(args=args, tool_context=tool_context_instance)
  assert result == expected_result


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "resource_path, expected_mime, fake_content",
    [
        ("references/doc.pdf", "application/pdf", b"fake pdf content"),
        ("assets/image.png", "image/png", b"fake image content"),
    ],
)
async def test_load_resource_process_llm_request_binary(
    mock_skill1,
    tool_context_instance,
    resource_path,
    expected_mime,
    fake_content,
):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.LoadSkillResourceTool(toolset)

  llm_req = mock.create_autospec(llm_request_model.LlmRequest, instance=True)

  part = types.Part.from_function_response(
      name=tool.name,
      response={
          "skill_name": "skill1",
          "file_path": resource_path,
          "status": (
              "Binary file detected. The content has been injected into the"
              " conversation history for you to analyze."
          ),
      },
  )
  content = types.Content(role="model", parts=[part])
  llm_req.contents = [content]

  await tool.process_llm_request(
      tool_context=tool_context_instance, llm_request=llm_req
  )

  assert len(llm_req.contents) == 2
  injected_content = llm_req.contents[1]
  assert injected_content.role == "user"
  assert len(injected_content.parts) == 2
  assert (
      f"The content of binary file '{resource_path}' is:"
      in injected_content.parts[0].text
  )
  assert injected_content.parts[1].inline_data.data == fake_content
  assert injected_content.parts[1].inline_data.mime_type == expected_mime


@pytest.mark.asyncio
async def test_process_llm_request_with_list_skills_tool(
    mock_skill1, mock_skill2, tool_context_instance
):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  llm_req = mock.create_autospec(llm_request_model.LlmRequest, instance=True)

  await toolset.process_llm_request(
      tool_context=tool_context_instance, llm_request=llm_req
  )

  llm_req.append_instructions.assert_called_once_with(
      [skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION]
  )


@pytest.mark.asyncio
async def test_process_llm_request_without_list_skills_tool(
    mock_skill1, mock_skill2, tool_context_instance
):
  toolset = skill_toolset.SkillToolset([mock_skill1, mock_skill2])
  # Manually remove ListSkillsTool from self._tools to simulate it not being available
  toolset._tools = [
      t
      for t in toolset._tools
      if not isinstance(t, skill_toolset.ListSkillsTool)
  ]

  llm_req = mock.create_autospec(llm_request_model.LlmRequest, instance=True)

  await toolset.process_llm_request(
      tool_context=tool_context_instance, llm_request=llm_req
  )

  llm_req.append_instructions.assert_called_once()
  args, _ = llm_req.append_instructions.call_args
  instructions = args[0]
  assert len(instructions) == 2
  assert instructions[0] == skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION
  assert "<available_skills>" in instructions[1]
  assert "skill1" in instructions[1]
  assert "skill2" in instructions[1]


def test_default_skill_system_instruction_warning():
  with pytest.warns(
      UserWarning, match="DEFAULT_SKILL_SYSTEM_INSTRUCTION is experimental"
  ):
    instruction = skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION
    assert "specialized 'skills'" in instruction


def test_duplicate_skill_name_raises(mock_skill1):
  skill_dup = mock.create_autospec(models.Skill, instance=True)
  skill_dup.name = "skill1"
  with pytest.raises(ValueError, match="Duplicate skill name"):
    skill_toolset.SkillToolset([mock_skill1, skill_dup])


@pytest.mark.asyncio
async def test_scripts_resource_not_found(mock_skill1, tool_context_instance):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.LoadSkillResourceTool(toolset)
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "scripts/nonexistent.sh"},
      tool_context=tool_context_instance,
  )
  assert result["error_code"] == "RESOURCE_NOT_FOUND"


# RunSkillScriptTool tests


def _make_tool_context_with_agent(agent=None):
  """Creates a mock ToolContext with _invocation_context.agent."""
  ctx = mock.MagicMock(spec=tool_context.ToolContext)
  ctx._invocation_context = mock.MagicMock()
  ctx._invocation_context.agent = agent or mock.MagicMock()
  ctx._invocation_context.agent.name = "test_agent"
  ctx._invocation_context.agent_states = {}
  ctx.agent_name = "test_agent"
  ctx.state = {}
  return ctx


def _make_mock_executor(stdout="", stderr=""):
  """Creates a mock code executor that returns the given output."""
  executor = mock.create_autospec(BaseCodeExecutor, instance=True)
  executor.execute_code.return_value = CodeExecutionResult(
      stdout=stdout, stderr=stderr
  )
  return executor


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "args, expected_error_code",
    [
        (
            {"file_path": "setup.sh"},
            "INVALID_ARGUMENTS",
        ),
        (
            {"skill_name": "skill1"},
            "INVALID_ARGUMENTS",
        ),
        (
            {"skill_name": "", "file_path": "setup.sh"},
            "INVALID_ARGUMENTS",
        ),
        (
            {"skill_name": "skill1", "file_path": ""},
            "INVALID_ARGUMENTS",
        ),
    ],
)
async def test_execute_script_missing_params(
    mock_skill1, args, expected_error_code
):
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(args=args, tool_context=ctx)
  assert result["error_code"] == expected_error_code


@pytest.mark.asyncio
async def test_execute_script_skill_not_found(mock_skill1):
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "nonexistent", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["error_code"] == "SKILL_NOT_FOUND"


@pytest.mark.asyncio
async def test_execute_script_script_not_found(mock_skill1):
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "nonexistent.py"},
      tool_context=ctx,
  )
  assert result["error_code"] == "SCRIPT_NOT_FOUND"


@pytest.mark.asyncio
async def test_execute_script_no_code_executor(mock_skill1):
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  # Agent without code_executor attribute
  agent = mock.MagicMock(spec=[])
  ctx = _make_tool_context_with_agent(agent=agent)
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["error_code"] == "NO_CODE_EXECUTOR"


@pytest.mark.asyncio
async def test_execute_script_agent_code_executor_none(mock_skill1):
  """Agent has code_executor attr but it's None."""
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  agent = mock.MagicMock()
  agent.code_executor = None
  ctx = _make_tool_context_with_agent(agent=agent)
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["error_code"] == "NO_CODE_EXECUTOR"


@pytest.mark.asyncio
async def test_execute_script_unsupported_type(mock_skill1):
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "build.rb"},
      tool_context=ctx,
  )
  assert result["error_code"] == "UNSUPPORTED_SCRIPT_TYPE"


@pytest.mark.asyncio
async def test_execute_script_python_success(mock_skill1):
  executor = _make_mock_executor(stdout="hello\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["status"] == "success"
  assert result["stdout"] == "hello\n"
  assert result["stderr"] == ""
  assert result["skill_name"] == "skill1"
  assert result["file_path"] == "run.py"

  # Verify the code passed to executor runs the python scripts
  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert "_materialize_and_run()" in code_input.code
  assert "import runpy" in code_input.code
  assert "sys.argv = ['scripts/run.py']" in code_input.code
  assert (
      "runpy.run_path('scripts/run.py', run_name='__main__')" in code_input.code
  )


@pytest.mark.asyncio
async def test_execute_script_shell_success(mock_skill1):
  executor = _make_mock_executor(stdout="setup\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "success"
  assert result["stdout"] == "setup\n"

  # Verify the code wraps in subprocess.run with JSON envelope
  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert "subprocess.run" in code_input.code
  assert "bash" in code_input.code
  assert "__shell_result__" in code_input.code


@pytest.mark.asyncio
async def test_execute_script_with_input_args_python(mock_skill1):
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "args": {"verbose": True, "count": "3"},
      },
      tool_context=ctx,
  )
  assert result["status"] == "success"

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert (
      "['scripts/run.py', '--verbose', 'True', '--count', '3']"
      in code_input.code
  )


@pytest.mark.asyncio
async def test_execute_script_with_input_args_shell(mock_skill1):
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "setup.sh",
          "args": {"force": True},
      },
      tool_context=ctx,
  )
  assert result["status"] == "success"

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert "['bash', 'scripts/setup.sh', '--force', 'True']" in code_input.code


@pytest.mark.asyncio
async def test_execute_script_with_list_args_python(
    mock_skill1,
):
  """Verifies that python scripts can be executed with list arguments."""
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "args": ["--verbose", "True", "-n", "5", "input.txt"],
      },
      tool_context=ctx,
  )
  assert result["status"] == "success"

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert (
      "['scripts/run.py', '--verbose', 'True', '-n', '5', 'input.txt']"
      in code_input.code
  )


@pytest.mark.asyncio
async def test_execute_script_with_list_args_shell(
    mock_skill1,
):
  """Verifies that shell scripts can be executed with list arguments."""
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "setup.sh",
          "args": ["-n", "5", "input.txt"],
      },
      tool_context=ctx,
  )
  assert result["status"] == "success"

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert (
      "['bash', 'scripts/setup.sh', '-n', '5', 'input.txt']" in code_input.code
  )


@pytest.mark.asyncio
async def test_execute_script_with_list_args_rejects_others_python(
    mock_skill1,  # pylint: disable=redefined-outer-name
):
  """Verifies that short_options and positional_args are rejected when args is a list for Python scripts."""
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "args": ["arg1", "arg2"],
          "short_options": {"v": True},
          "positional_args": ["pos1"],
      },
      tool_context=ctx,
  )
  assert result["error_code"] == "INVALID_ARGUMENTS"
  assert (
      "Cannot specify 'short_options' or 'positional_args'" in result["error"]
  )


@pytest.mark.asyncio
async def test_execute_script_with_list_args_rejects_others_shell(
    mock_skill1,  # pylint: disable=redefined-outer-name
):
  """Verifies that short_options and positional_args are rejected when args is a list for shell scripts."""
  executor = _make_mock_executor(stdout="done\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "setup.sh",
          "args": ["arg1", "arg2"],
          "short_options": {"v": True},
          "positional_args": ["pos1"],
      },
      tool_context=ctx,
  )
  assert result["error_code"] == "INVALID_ARGUMENTS"
  assert (
      "Cannot specify 'short_options' or 'positional_args'" in result["error"]
  )


@pytest.mark.asyncio
async def test_execute_script_scripts_prefix_stripping(mock_skill1):
  executor = _make_mock_executor(stdout="setup\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "scripts/setup.sh",
      },
      tool_context=ctx,
  )
  assert result["status"] == "success"
  assert result["file_path"] == "scripts/setup.sh"


@pytest.mark.asyncio
async def test_execute_script_toolset_executor_priority(mock_skill1):
  """Toolset-level executor takes priority over agent's."""
  toolset_executor = _make_mock_executor(stdout="from toolset\n")
  agent_executor = _make_mock_executor(stdout="from agent\n")
  toolset = skill_toolset.SkillToolset(
      [mock_skill1], code_executor=toolset_executor
  )
  tool = skill_toolset.RunSkillScriptTool(toolset)
  agent = mock.MagicMock()
  agent.code_executor = agent_executor
  ctx = _make_tool_context_with_agent(agent=agent)
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["stdout"] == "from toolset\n"
  toolset_executor.execute_code.assert_called_once()
  agent_executor.execute_code.assert_not_called()


@pytest.mark.asyncio
async def test_execute_script_agent_executor_fallback(mock_skill1):
  """Falls back to agent's code executor when toolset has none."""
  agent_executor = _make_mock_executor(stdout="from agent\n")
  toolset = skill_toolset.SkillToolset([mock_skill1])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  agent = mock.MagicMock()
  agent.code_executor = agent_executor
  ctx = _make_tool_context_with_agent(agent=agent)
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["stdout"] == "from agent\n"
  agent_executor.execute_code.assert_called_once()


@pytest.mark.asyncio
async def test_execute_script_execution_error(mock_skill1):
  executor = _make_mock_executor()
  executor.execute_code.side_effect = RuntimeError("boom")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["error_code"] == "EXECUTION_ERROR"
  assert "boom" in result["error"]
  assert result["error"].startswith("Failed to execute script 'run.py':")


@pytest.mark.asyncio
async def test_execute_script_stderr_only_sets_error_status(mock_skill1):
  """stderr with no stdout should report error status."""
  executor = _make_mock_executor(stdout="", stderr="fatal error\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["status"] == "error"
  assert result["stderr"] == "fatal error\n"


@pytest.mark.asyncio
async def test_execute_script_stderr_with_stdout_sets_warning(mock_skill1):
  """stderr alongside stdout should report warning status."""
  executor = _make_mock_executor(stdout="output\n", stderr="deprecation\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["status"] == "warning"
  assert result["stdout"] == "output\n"
  assert result["stderr"] == "deprecation\n"


@pytest.mark.asyncio
async def test_execute_script_execution_error_truncated(mock_skill1):
  """Long exception messages are truncated to avoid wasting LLM tokens."""
  executor = _make_mock_executor()
  executor.execute_code.side_effect = RuntimeError("x" * 300)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["error_code"] == "EXECUTION_ERROR"
  # 200 chars of the message + "..." suffix + the prefix
  assert result["error"].endswith("...")
  assert len(result["error"]) < 300


@pytest.mark.asyncio
async def test_execute_script_system_exit_caught(mock_skill1):
  """sys.exit() in a script should not terminate the process."""
  executor = _make_mock_executor()
  executor.execute_code.side_effect = SystemExit(1)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["error_code"] == "EXECUTION_ERROR"
  assert "exited with code 1" in result["error"]


@pytest.mark.asyncio
async def test_execute_script_system_exit_zero_is_success(mock_skill1):
  """sys.exit(0) is a normal termination and should report success."""
  executor = _make_mock_executor()
  executor.execute_code.side_effect = SystemExit(0)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()

  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["status"] == "success"


@pytest.mark.asyncio
async def test_execute_script_system_exit_none_is_success(mock_skill1):
  """sys.exit() with no arg (None) should report success."""
  executor = _make_mock_executor()
  executor.execute_code.side_effect = SystemExit(None)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )
  assert result["status"] == "success"


@pytest.mark.asyncio
async def test_execute_script_shell_includes_timeout(mock_skill1):
  """Shell wrapper includes timeout in subprocess.run."""
  executor = _make_mock_executor(stdout="ok\n")
  toolset = skill_toolset.SkillToolset(
      [mock_skill1], code_executor=executor, script_timeout=60
  )
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "success"
  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert "timeout=60" in code_input.code


@pytest.mark.asyncio
async def test_execute_script_extensionless_unsupported(mock_skill1):
  """Files without extensions should return UNSUPPORTED_SCRIPT_TYPE."""
  # Add a script with no extension to the mock
  original_side_effect = mock_skill1.resources.get_script.side_effect

  def get_script_extended(name):
    if name == "noext":
      return models.Script(src="print('hi')")
    return original_side_effect(name)

  mock_skill1.resources.get_script.side_effect = get_script_extended

  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "noext"},
      tool_context=ctx,
  )
  assert result["error_code"] == "UNSUPPORTED_SCRIPT_TYPE"


# ── Integration tests using real UnsafeLocalCodeExecutor ──


def _make_skill_with_script(skill_name, script_name, script):
  """Creates a minimal mock Skill with a single script."""
  skill = mock.create_autospec(models.Skill, instance=True)
  skill.name = skill_name
  skill.description = f"Test skill {skill_name}"
  skill.instructions = "test instructions"
  fm = mock.create_autospec(models.Frontmatter, instance=True)
  fm.name = skill_name
  fm.description = f"Test skill {skill_name}"
  skill.frontmatter = fm
  skill.resources = mock.MagicMock(
      spec=[
          "get_reference",
          "get_asset",
          "get_script",
          "list_references",
          "list_assets",
          "list_scripts",
      ]
  )

  def get_script(name):
    if name == script_name:
      return script
    return None

  skill.resources.get_script.side_effect = get_script
  skill.resources.get_reference.return_value = None
  skill.resources.get_asset.return_value = None
  skill.resources.list_references.return_value = []
  skill.resources.list_assets.return_value = []
  skill.resources.list_scripts.return_value = [script_name]
  return skill


def _make_real_executor_toolset(skills, **kwargs):
  """Creates a SkillToolset with a real UnsafeLocalCodeExecutor."""

  if sys.executable is None:
    sys.executable = "/usr/bin/python3"

  executor = UnsafeLocalCodeExecutor(timeout_seconds=10)
  return skill_toolset.SkillToolset(skills, code_executor=executor, **kwargs)


@pytest.mark.asyncio
async def test_integration_python_stdout():
  """Real executor: Python script stdout is captured."""
  script = models.Script(src="print('hello world')")
  skill = _make_skill_with_script("test_skill", "hello.py", script)
  toolset = _make_real_executor_toolset([skill])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "test_skill",
          "file_path": "hello.py",
      },
      tool_context=ctx,
  )
  assert "status" in result, f"Result missing status: {result}"
  assert result["status"] == "success"
  assert result["stdout"] == "hello world\n"
  assert result["stderr"] == ""


@pytest.mark.asyncio
async def test_integration_python_sys_exit_zero():
  """Real executor: sys.exit(0) is treated as success."""
  script = models.Script(src="import sys; sys.exit(0)")
  skill = _make_skill_with_script("test_skill", "exit_zero.py", script)
  toolset = _make_real_executor_toolset([skill])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "test_skill",
          "file_path": "exit_zero.py",
      },
      tool_context=ctx,
  )
  assert "status" in result, f"Result missing status: {result}"
  assert result["status"] == "success"


@pytest.mark.asyncio
async def test_integration_shell_stdout_and_stderr():
  """Real executor: shell script preserves both stdout and stderr."""
  script = models.Script(src="echo output; echo warning >&2")
  skill = _make_skill_with_script("test_skill", "both.sh", script)
  toolset = _make_real_executor_toolset([skill])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "test_skill",
          "file_path": "both.sh",
      },
      tool_context=ctx,
  )
  assert "status" in result, f"Result missing status: {result}"
  assert result["status"] == "warning"
  assert "output" in result["stdout"]
  assert "warning" in result["stderr"]


@pytest.mark.asyncio
async def test_integration_shell_stderr_only():
  """Real executor: shell script with only stderr reports error."""
  script = models.Script(src="echo failure >&2")
  skill = _make_skill_with_script("test_skill", "err.sh", script)
  toolset = _make_real_executor_toolset([skill])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "test_skill",
          "file_path": "err.sh",
      },
      tool_context=ctx,
  )
  assert "status" in result, f"Result missing status: {result}"
  assert result["status"] == "error"
  assert "failure" in result["stderr"]


# ── Shell JSON envelope parsing (unit tests with mock executor) ──


@pytest.mark.asyncio
async def test_shell_json_envelope_parsed(mock_skill1):
  """Shell JSON envelope is correctly unpacked by run_async."""
  import json

  envelope = json.dumps({
      "__shell_result__": True,
      "stdout": "hello from shell\n",
      "stderr": "",
      "returncode": 0,
  })
  executor = _make_mock_executor(stdout=envelope)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "success"
  assert result["stdout"] == "hello from shell\n"
  assert result["stderr"] == ""


@pytest.mark.asyncio
async def test_shell_json_envelope_nonzero_returncode(mock_skill1):
  """Non-zero returncode in shell envelope sets stderr."""
  import json

  envelope = json.dumps({
      "__shell_result__": True,
      "stdout": "",
      "stderr": "",
      "returncode": 2,
  })
  executor = _make_mock_executor(stdout=envelope)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "error"
  assert "Exit code 2" in result["stderr"]


@pytest.mark.asyncio
async def test_shell_json_envelope_with_stderr(mock_skill1):
  """Shell envelope with both stdout and stderr reports warning."""
  import json

  envelope = json.dumps({
      "__shell_result__": True,
      "stdout": "data\n",
      "stderr": "deprecation warning\n",
      "returncode": 0,
  })
  executor = _make_mock_executor(stdout=envelope)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "warning"
  assert result["stdout"] == "data\n"
  assert result["stderr"] == "deprecation warning\n"


@pytest.mark.asyncio
async def test_shell_json_envelope_timeout(mock_skill1):
  """Shell envelope from TimeoutExpired reports error status."""
  import json

  envelope = json.dumps({
      "__shell_result__": True,
      "stdout": "partial output\n",
      "stderr": "Timed out after 300s",
      "returncode": -1,
  })
  executor = _make_mock_executor(stdout=envelope)
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "error"
  assert result["stdout"] == "partial output\n"
  assert "Timed out" in result["stderr"]


@pytest.mark.asyncio
async def test_shell_non_json_stdout_passthrough(mock_skill1):
  """Non-JSON shell stdout is passed through without parsing."""
  executor = _make_mock_executor(stdout="plain text output\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={"skill_name": "skill1", "file_path": "setup.sh"},
      tool_context=ctx,
  )
  assert result["status"] == "success"
  assert result["stdout"] == "plain text output\n"


# ── input_files packaging ──


@pytest.mark.asyncio
async def test_execute_script_input_files_packaged(mock_skill1):
  """Verify references, assets, and scripts are packaged inside the wrapper code."""
  executor = _make_mock_executor(stdout="ok\n")
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  await tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=ctx,
  )

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]

  # input_files is no longer populated; it's serialized inside the script
  assert code_input.input_files is None or len(code_input.input_files) == 0

  # Ensure the extracted literal contains our fake files
  assert "references/ref1.md" in code_input.code
  assert "assets/asset1.txt" in code_input.code
  assert "scripts/setup.sh" in code_input.code
  assert "scripts/run.py" in code_input.code
  assert "scripts/build.rb" in code_input.code

  # Verify content mappings exist in the string
  assert "'references/ref1.md': 'ref content 1'" in code_input.code
  assert "'assets/asset1.txt': 'asset content 1'" in code_input.code


# ── Integration: shell non-zero exit ──


@pytest.mark.asyncio
async def test_integration_shell_nonzero_exit():
  """Real executor: shell script with non-zero exit via JSON envelope."""
  script = models.Script(src="exit 42")
  skill = _make_skill_with_script("test_skill", "fail.sh", script)
  toolset = _make_real_executor_toolset([skill])
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "test_skill",
          "file_path": "fail.sh",
      },
      tool_context=ctx,
  )
  assert "status" in result, f"Result missing status: {result}"
  assert result["status"] == "error"
  assert "42" in result["stderr"]


# ── Finding 1: system instruction references correct tool name ──


def test_system_instruction_references_run_skill_script():
  """System instruction must reference the actual tool name."""
  assert "run_skill_script" in skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION
  assert (
      "execute_skill_script"
      not in skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION
  )


# ── Finding 2: empty files are mounted (not silently dropped) ──


@pytest.mark.asyncio
async def test_execute_script_empty_files_mounted():
  """Verify empty files are included in wrapper code, not dropped."""
  skill = mock.create_autospec(models.Skill, instance=True)
  skill.name = "skill_empty"
  skill.resources = mock.MagicMock(
      spec=[
          "get_reference",
          "get_asset",
          "get_script",
          "list_references",
          "list_assets",
          "list_scripts",
      ]
  )
  skill.resources.get_reference.side_effect = (
      lambda n: "" if n == "empty.md" else None
  )
  skill.resources.get_asset.side_effect = (
      lambda n: "" if n == "empty.cfg" else None
  )
  skill.resources.get_script.side_effect = (
      lambda n: models.Script(src="") if n == "run.py" else None
  )
  skill.resources.list_references.return_value = ["empty.md"]
  skill.resources.list_assets.return_value = ["empty.cfg"]
  skill.resources.list_scripts.return_value = ["run.py"]

  executor = _make_mock_executor(stdout="ok\n")
  toolset = skill_toolset.SkillToolset([skill], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  await tool.run_async(
      args={"skill_name": "skill_empty", "file_path": "run.py"},
      tool_context=ctx,
  )

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  assert "'references/empty.md': ''" in code_input.code
  assert "'assets/empty.cfg': ''" in code_input.code
  assert "'scripts/run.py': ''" in code_input.code


# ── Finding 3: invalid args type returns clear error ──


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "bad_args",
    [
        "not a dict",
        42,
        True,
    ],
)
async def test_execute_script_invalid_args_type(mock_skill1, bad_args):
  """Non-dict args should return INVALID_ARGS_TYPE, not crash."""
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "args": bad_args,
      },
      tool_context=ctx,
  )
  assert result["error_code"] == "INVALID_ARGUMENTS"
  executor.execute_code.assert_not_called()


@pytest.mark.parametrize(
    "bad_short_options",
    [
        "not a dict",
        42,
        True,
        ["list"],
    ],
)
@pytest.mark.asyncio
async def test_execute_script_invalid_short_options_type(
    mock_skill1, bad_short_options
):
  """Non-dict short_options should return INVALID_SHORT_OPTIONS_TYPE, not crash."""
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "short_options": bad_short_options,
      },
      tool_context=ctx,
  )
  assert result["error_code"] == "INVALID_ARGUMENTS"
  executor.execute_code.assert_not_called()


@pytest.mark.parametrize(
    "bad_positional_args",
    [
        "not a list",
        42,
        True,
        {"dict": 1},
    ],
)
@pytest.mark.asyncio
async def test_execute_script_invalid_positional_args_type(
    mock_skill1, bad_positional_args
):
  """Non-list positional_args should return INVALID_POSITIONAL_ARGS_TYPE, not crash."""
  executor = _make_mock_executor()
  toolset = skill_toolset.SkillToolset([mock_skill1], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  result = await tool.run_async(
      args={
          "skill_name": "skill1",
          "file_path": "run.py",
          "positional_args": bad_positional_args,
      },
      tool_context=ctx,
  )
  assert result["error_code"] == "INVALID_ARGUMENTS"
  executor.execute_code.assert_not_called()


# ── Finding 4: binary file content is handled in wrapper ──


@pytest.mark.asyncio
async def test_execute_script_binary_content_packaged():
  """Verify binary asset content uses 'wb' mode in wrapper code."""
  skill = mock.create_autospec(models.Skill, instance=True)
  skill.name = "skill_bin"
  skill.resources = mock.MagicMock(
      spec=[
          "get_reference",
          "get_asset",
          "get_script",
          "list_references",
          "list_assets",
          "list_scripts",
      ]
  )
  skill.resources.get_reference.side_effect = (
      lambda n: b"\x00\x01\x02" if n == "data.bin" else None
  )
  skill.resources.get_asset.return_value = None
  skill.resources.get_script.side_effect = lambda n: (
      models.Script(src="print('ok')") if n == "run.py" else None
  )
  skill.resources.list_references.return_value = ["data.bin"]
  skill.resources.list_assets.return_value = []
  skill.resources.list_scripts.return_value = ["run.py"]

  executor = _make_mock_executor(stdout="ok\n")
  toolset = skill_toolset.SkillToolset([skill], code_executor=executor)
  tool = skill_toolset.RunSkillScriptTool(toolset)
  ctx = _make_tool_context_with_agent()
  await tool.run_async(
      args={"skill_name": "skill_bin", "file_path": "run.py"},
      tool_context=ctx,
  )

  call_args = executor.execute_code.call_args
  code_input = call_args[0][1]
  # Binary content should appear as bytes literal
  assert "b'\\x00\\x01\\x02'" in code_input.code
  # Wrapper code handles binary with 'wb' mode
  assert "'wb' if isinstance(content, bytes)" in code_input.code


@pytest.mark.asyncio
async def test_skill_toolset_dynamic_tool_resolution(mock_skill1, mock_skill2):
  # Set up skills with additional_tools in metadata
  mock_skill1.frontmatter.metadata = {
      "adk_additional_tools": ["my_custom_tool", "my_func", "shared_tool"]
  }
  mock_skill1.name = "skill1"

  mock_skill2.frontmatter.metadata = {
      "adk_additional_tools": [
          "skill2_tool",
          "shared_tool",
          "prefixed_mock_tool",
      ]
  }
  mock_skill2.name = "skill2"

  # Prepare additional tools
  custom_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  custom_tool.name = "my_custom_tool"

  skill2_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  skill2_tool.name = "skill2_tool"

  shared_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  shared_tool.name = "shared_tool"

  def my_func():
    """My function description."""
    pass

  # Setup prefixed toolset
  mock_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  mock_tool.name = "prefixed_mock_tool"
  prefixed_set = mock.create_autospec(skill_toolset.BaseToolset, instance=True)
  prefixed_set.get_tools_with_prefix.return_value = [mock_tool]

  toolset = skill_toolset.SkillToolset(
      [mock_skill1, mock_skill2],
      additional_tools=[
          custom_tool,
          skill2_tool,
          shared_tool,
          my_func,
          prefixed_set,
      ],
  )

  ctx = _make_tool_context_with_agent()
  ctx.invocation_id = "turn-1"
  # Initial tools (only core)
  tools1 = await toolset.get_tools_with_prefix(readonly_context=ctx)
  assert len(tools1) == 4

  # Activate skills
  load_tool = skill_toolset.LoadSkillTool(toolset)
  await load_tool.run_async(args={"skill_name": "skill1"}, tool_context=ctx)
  await load_tool.run_async(args={"skill_name": "skill2"}, tool_context=ctx)

  # Dynamic tools should now be resolved
  ctx.invocation_id = "turn-2"
  tools = await toolset.get_tools_with_prefix(readonly_context=ctx)
  assert tools is not tools1
  tool_names = {t.name for t in tools}

  # Core tools
  assert "list_skills" in tool_names
  assert "load_skill" in tool_names
  assert "load_skill_resource" in tool_names
  assert "run_skill_script" in tool_names

  # Skill 1 tools
  assert "my_custom_tool" in tool_names
  assert "my_func" in tool_names

  # Skill 2 tools
  assert "skill2_tool" in tool_names

  # Shared tool (should only appear once)
  assert "shared_tool" in tool_names
  assert len([t for t in tools if t.name == "shared_tool"]) == 1

  # Prefixed toolset tool
  assert "prefixed_mock_tool" in tool_names

  # Check specific tool resolution details
  my_func_tool = next(t for t in tools if t.name == "my_func")
  assert isinstance(my_func_tool, skill_toolset.FunctionTool)
  assert my_func_tool.description == "My function description."


@pytest.mark.asyncio
async def test_skill_toolset_resolution_error_handling(mock_skill1, caplog):
  mock_skill1.frontmatter.metadata = {
      "adk_additional_tools": ["nonexistent_tool"]
  }
  mock_skill1.name = "skill1"
  toolset = skill_toolset.SkillToolset([mock_skill1])
  ctx = _make_tool_context_with_agent()

  # Activate skill
  load_tool = skill_toolset.LoadSkillTool(toolset)
  await load_tool.run_async(args={"skill_name": "skill1"}, tool_context=ctx)

  with caplog.at_level(logging.WARNING):
    tools = await toolset.get_tools(readonly_context=ctx)

  # Should still return basic skill tools
  assert len(tools) == 4


@pytest.fixture(name="mock_registry")
def _mock_registry():
  """Fixture for mock SkillRegistry."""
  registry = mock.create_autospec(skill_toolset.SkillRegistry, instance=True)
  registry.search_tool_description.return_value = None
  return registry


@pytest.mark.asyncio
async def test_skill_toolset_init_with_registry(mock_registry):
  # Verify toolset initializes with empty skills list and registers SearchSkillsTool
  toolset = skill_toolset.SkillToolset(registry=mock_registry)
  assert toolset._registry == mock_registry
  assert len(toolset._skills) == 0

  tools = await toolset.get_tools()
  assert len(tools) == 5
  assert isinstance(tools[4], skill_toolset.SearchSkillsTool)


def test_search_skills_tool_init_without_registry():
  toolset = skill_toolset.SkillToolset()
  with pytest.raises(
      ValueError,
      match="SearchSkillsTool requires a configured skill registry.",
  ):
    skill_toolset.SearchSkillsTool(toolset)


@pytest.mark.asyncio
async def test_search_skills_tool_run_async(
    mock_registry, mock_skill1, tool_context_instance
):
  # Verify search_skills tool works, filters out local naming conflicts
  mock_frontmatter1 = mock.create_autospec(models.Frontmatter, instance=True)
  mock_frontmatter1.name = "skill1"
  mock_frontmatter1.model_dump.return_value = {"name": "skill1"}

  mock_frontmatter2 = mock.create_autospec(models.Frontmatter, instance=True)
  mock_frontmatter2.name = "skill2"
  mock_frontmatter2.model_dump.return_value = {"name": "skill2"}

  mock_registry.search_skills.return_value = [
      mock_frontmatter1,
      mock_frontmatter2,
  ]

  # skill1 exists locally, skill2 does not
  toolset = skill_toolset.SkillToolset([mock_skill1], registry=mock_registry)
  tool = skill_toolset.SearchSkillsTool(toolset)

  result = await tool.run_async(
      args={"query": "test"}, tool_context=tool_context_instance
  )

  mock_registry.search_skills.assert_called_once_with(query="test")
  # skill1 should be filtered out due to naming conflict with local mock_skill1
  assert result == [{"name": "skill2"}]


@pytest.mark.asyncio
async def test_load_skill_tool_fallback_to_registry(
    mock_registry, mock_skill1, tool_context_instance
):
  # Verify LoadSkillTool falls back to registry, fetching on-demand without cache
  mock_registry.get_skill.return_value = mock_skill1

  toolset = skill_toolset.SkillToolset(registry=mock_registry)
  tool = skill_toolset.LoadSkillTool(toolset)

  # Mock state
  state_key = "_adk_activated_skill_test_agent"
  tool_context_instance.state.get.return_value = None

  # First load: goes to registry
  tool_context_instance.invocation_id = "inv-1"
  result = await tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )

  assert result["skill_name"] == "skill1"
  assert result["instructions"] == "instructions for skill1"
  mock_registry.get_skill.assert_called_once_with(name="skill1")

  # Verify that the Skill frontmatter was cached in the unified state dictionary
  tool_context_instance.state.__setitem__.assert_called_once_with(
      state_key, ["skill1"]
  )

  # Mock state.get to return the cached skill list
  tool_context_instance.state.get.side_effect = lambda key, default=None: (
      ["skill1"] if key == state_key else default
  )

  # Second load on a new turn: should fetch from registry on-demand as only frontmatter is in state
  tool_context_instance.invocation_id = "inv-2"
  mock_registry.get_skill.reset_mock()
  result2 = await tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )
  assert result2["skill_name"] == "skill1"
  mock_registry.get_skill.assert_called_once_with(name="skill1")


@pytest.mark.asyncio
async def test_registry_skill_resources_and_tools_resolved(
    mock_registry, tool_context_instance
):
  # Create a mock registry skill that declares local additional tools
  mock_skill = mock.create_autospec(models.Skill, instance=True)
  mock_skill.name = "registry_skill"
  mock_skill.instructions = "registry instructions"
  mock_skill.frontmatter = mock.create_autospec(
      models.Frontmatter, instance=True
  )
  mock_skill.frontmatter.name = "registry_skill"
  mock_skill.frontmatter.metadata = {"adk_additional_tools": ["my_custom_tool"]}

  mock_skill.resources = mock.MagicMock()
  mock_skill.resources.get_reference.return_value = "reference content"

  mock_registry.get_skill.return_value = mock_skill

  # Setup toolset with the registry and the local implementation of the tool
  custom_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  custom_tool.name = "my_custom_tool"

  toolset = skill_toolset.SkillToolset(
      registry=mock_registry, additional_tools=[custom_tool]
  )

  # Load the skill via LoadSkillTool
  load_tool = skill_toolset.LoadSkillTool(toolset)

  state_key = "_adk_activated_skill_test_agent"
  tool_context_instance.state.get.side_effect = lambda key, default=None: (
      ["registry_skill"] if key == state_key else default
  )

  result = await load_tool.run_async(
      args={"skill_name": "registry_skill"}, tool_context=tool_context_instance
  )
  assert result["skill_name"] == "registry_skill"

  # 1. Verify dynamic tools from the registry are resolved
  tools = await toolset.get_tools(readonly_context=tool_context_instance)
  tool_names = {t.name for t in tools}
  assert "my_custom_tool" in tool_names

  # 2. Verify resource loading resolves registry skill correctly
  resource_tool = skill_toolset.LoadSkillResourceTool(toolset)
  res_result = await resource_tool.run_async(
      args={
          "skill_name": "registry_skill",
          "file_path": "references/ref.md",
      },
      tool_context=tool_context_instance,
  )
  assert res_result["content"] == "reference content"


@pytest.mark.asyncio
async def test_process_llm_request_with_registry(
    mock_registry, tool_context_instance
):
  toolset = skill_toolset.SkillToolset(registry=mock_registry)
  llm_req = mock.create_autospec(llm_request_model.LlmRequest, instance=True)

  await toolset.process_llm_request(
      tool_context=tool_context_instance, llm_request=llm_req
  )

  llm_req.append_instructions.assert_called_once()
  args, _ = llm_req.append_instructions.call_args
  instructions = args[0]
  assert len(instructions) == 2
  assert instructions[0] == skill_toolset.DEFAULT_SKILL_SYSTEM_INSTRUCTION
  assert "search_skills" in instructions[1]


@pytest.mark.asyncio
async def test_turn_scoped_skill_cache(
    mock_registry, mock_skill1, tool_context_instance
):
  # Verify that multiple tool calls on the same registry-provided skill in the same turn
  # use the turn-scoped cache and only call the registry once.
  mock_registry.get_skill.return_value = mock_skill1

  toolset = skill_toolset.SkillToolset(registry=mock_registry)
  load_tool = skill_toolset.LoadSkillTool(toolset)
  script_tool = skill_toolset.RunSkillScriptTool(toolset)

  tool_context_instance.invocation_id = "same-turn-id"
  tool_context_instance.state.get.return_value = None

  # Call LoadSkillTool
  res1 = await load_tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )
  assert res1["skill_name"] == "skill1"
  mock_registry.get_skill.assert_called_once_with(name="skill1")

  # Setup executor for script tool
  executor = mock.create_autospec(skill_toolset.BaseCodeExecutor, instance=True)
  executor.execute_code.return_value = mock.MagicMock(
      stdout="hello\n", stderr=""
  )
  toolset._code_executor = executor

  # Call RunSkillScriptTool in the same turn
  res2 = await script_tool.run_async(
      args={"skill_name": "skill1", "file_path": "run.py"},
      tool_context=tool_context_instance,
  )
  assert res2["status"] == "success"
  # Registry should NOT be called again
  mock_registry.get_skill.assert_called_once_with(name="skill1")


@pytest.mark.asyncio
async def test_turn_scoped_skill_cache_eviction(mock_registry, mock_skill1):
  mock_registry.get_skill.return_value = mock_skill1
  toolset = skill_toolset.SkillToolset(registry=mock_registry)

  # Fill cache up to limit
  for i in range(16):
    await toolset._get_or_fetch_skill("skill1", f"turn-{i}")

  assert len(toolset._fetched_skill_cache) == 16
  assert "turn-0" in toolset._fetched_skill_cache

  # Next turn should evict oldest (turn-0)
  await toolset._get_or_fetch_skill("skill1", "turn-16")
  assert len(toolset._fetched_skill_cache) == 16
  assert "turn-0" not in toolset._fetched_skill_cache
  assert "turn-1" in toolset._fetched_skill_cache


@pytest.mark.asyncio
async def test_turn_scoped_skill_cache_concurrency(mock_registry, mock_skill1):
  # Delay registry fetch to simulate async I/O and force race condition
  async def delayed_get_skill(name):
    await asyncio.sleep(0.1)
    return mock_skill1

  mock_registry.get_skill.side_effect = delayed_get_skill
  toolset = skill_toolset.SkillToolset(registry=mock_registry)

  # Trigger concurrent calls for the same skill in the same turn
  results = await asyncio.gather(
      toolset._get_or_fetch_skill("skill1", "concurrent-turn"),
      toolset._get_or_fetch_skill("skill1", "concurrent-turn"),
      toolset._get_or_fetch_skill("skill1", "concurrent-turn"),
  )

  for res in results:
    assert res is mock_skill1

  # Registry should have been called exactly once
  mock_registry.get_skill.assert_called_once_with(name="skill1")


def test_skill_toolset_disables_invocation_cache():
  """Verify SkillToolset disables tool invocation caching to allow dynamic tools."""
  toolset = skill_toolset.SkillToolset()
  assert toolset._use_invocation_cache is False


@pytest.mark.asyncio
async def test_close_cancels_futures_and_clears_cache():
  # pylint: disable=protected-access
  toolset = skill_toolset.SkillToolset()

  # Create mock futures for testing close() behavior
  loop = asyncio.get_running_loop()
  fut1 = loop.create_future()
  fut2 = loop.create_future()
  fut2.set_result(None)  # Already done future

  toolset._fetched_skill_cache = collections.OrderedDict(
      {
          "turn1": {
              "skill1": fut1,
              "skill2": fut2,
          }
      }
  )

  await toolset.close()

  assert fut1.cancelled()
  assert not fut2.cancelled()  # Done futures shouldn't/can't be cancelled
  assert not toolset._fetched_skill_cache


# --- Tests for max_activated_skills (rolling window) ---


@pytest.mark.asyncio
async def test_max_activated_skills_evicts_oldest(
    mock_skill1, mock_skill2, tool_context_instance
):
  """When max_activated_skills is exceeded, the oldest skill is evicted."""
  toolset = skill_toolset.SkillToolset(
      [mock_skill1, mock_skill2], max_activated_skills=1
  )
  tool = skill_toolset.LoadSkillTool(toolset)

  state_key = "_adk_activated_skill_test_agent"

  # Load skill1
  tool_context_instance.state.get.return_value = None
  await tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )
  tool_context_instance.state.__setitem__.assert_called_with(
      state_key, ["skill1"]
  )

  # Load skill2 — skill1 should be evicted
  tool_context_instance.state.get.return_value = ["skill1"]
  await tool.run_async(
      args={"skill_name": "skill2"}, tool_context=tool_context_instance
  )
  tool_context_instance.state.__setitem__.assert_called_with(
      state_key, ["skill2"]
  )


@pytest.mark.asyncio
async def test_max_activated_skills_keeps_recent_n(
    tool_context_instance,
):
  """Rolling window keeps the N most recently activated skills."""
  skills = []
  for i in range(5):
    skill = mock.create_autospec(models.Skill, instance=True)
    skill.name = f"skill_{i}"
    skill.instructions = f"instructions for skill_{i}"
    frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
    frontmatter.name = f"skill_{i}"
    frontmatter.metadata = {}
    frontmatter.model_dump.return_value = {"name": f"skill_{i}"}
    skill.frontmatter = frontmatter
    skills.append(skill)

  toolset = skill_toolset.SkillToolset(skills, max_activated_skills=3)
  tool = skill_toolset.LoadSkillTool(toolset)

  state_key = "_adk_activated_skill_test_agent"

  # Sequentially load skills 0 through 4
  current_state = []
  for i in range(5):
    tool_context_instance.state.get.return_value = list(current_state)
    await tool.run_async(
        args={"skill_name": f"skill_{i}"}, tool_context=tool_context_instance
    )
    # Capture what was set
    call_args = tool_context_instance.state.__setitem__.call_args
    current_state = call_args[0][1]

  # After loading 5 skills with max=3, only the last 3 should remain
  assert current_state == ["skill_2", "skill_3", "skill_4"]


@pytest.mark.asyncio
async def test_max_activated_skills_reloading_moves_to_end(
    mock_skill1, mock_skill2, tool_context_instance
):
  """Re-loading an already active skill moves it to the end (most recent)."""
  # Create a third skill
  mock_skill3 = mock.create_autospec(models.Skill, instance=True)
  mock_skill3.name = "skill3"
  mock_skill3.instructions = "instructions for skill3"
  frontmatter3 = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter3.name = "skill3"
  frontmatter3.metadata = {}
  frontmatter3.model_dump.return_value = {"name": "skill3"}
  mock_skill3.frontmatter = frontmatter3

  toolset = skill_toolset.SkillToolset(
      [mock_skill1, mock_skill2, mock_skill3], max_activated_skills=3
  )
  tool = skill_toolset.LoadSkillTool(toolset)

  state_key = "_adk_activated_skill_test_agent"

  # Start with all 3 loaded
  tool_context_instance.state.get.return_value = [
      "skill1",
      "skill2",
      "skill3",
  ]

  # Reload skill1 — should move to end
  await tool.run_async(
      args={"skill_name": "skill1"}, tool_context=tool_context_instance
  )
  call_args = tool_context_instance.state.__setitem__.call_args
  assert call_args[0][1] == ["skill2", "skill3", "skill1"]


@pytest.mark.asyncio
async def test_no_max_activated_skills_unlimited(
    tool_context_instance,
):
  """Without max_activated_skills, all skills are retained (no eviction)."""
  skills = []
  for i in range(10):
    skill = mock.create_autospec(models.Skill, instance=True)
    skill.name = f"skill_{i}"
    skill.instructions = f"instructions for skill_{i}"
    frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
    frontmatter.name = f"skill_{i}"
    frontmatter.metadata = {}
    frontmatter.model_dump.return_value = {"name": f"skill_{i}"}
    skill.frontmatter = frontmatter
    skills.append(skill)

  toolset = skill_toolset.SkillToolset(skills)  # No max_activated_skills
  tool = skill_toolset.LoadSkillTool(toolset)

  state_key = "_adk_activated_skill_test_agent"

  current_state = []
  for i in range(10):
    tool_context_instance.state.get.return_value = list(current_state)
    await tool.run_async(
        args={"skill_name": f"skill_{i}"}, tool_context=tool_context_instance
    )
    call_args = tool_context_instance.state.__setitem__.call_args
    current_state = call_args[0][1]

  # All 10 skills should be retained
  assert len(current_state) == 10
  assert current_state == [f"skill_{i}" for i in range(10)]


@pytest.mark.asyncio
async def test_evicted_skill_tools_not_resolved():
  """After a skill is evicted, its additional tools are no longer resolved."""
  # Create two skills, each declaring different additional tools
  skill_a = mock.create_autospec(models.Skill, instance=True)
  skill_a.name = "skill_a"
  skill_a.instructions = "instructions A"
  frontmatter_a = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter_a.name = "skill_a"
  frontmatter_a.metadata = {"adk_additional_tools": ["tool_alpha"]}
  frontmatter_a.model_dump.return_value = {"name": "skill_a"}
  skill_a.frontmatter = frontmatter_a

  skill_b = mock.create_autospec(models.Skill, instance=True)
  skill_b.name = "skill_b"
  skill_b.instructions = "instructions B"
  frontmatter_b = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter_b.name = "skill_b"
  frontmatter_b.metadata = {"adk_additional_tools": ["tool_beta"]}
  frontmatter_b.model_dump.return_value = {"name": "skill_b"}
  skill_b.frontmatter = frontmatter_b

  # Create mock additional tools
  tool_alpha = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  tool_alpha.name = "tool_alpha"
  tool_beta = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  tool_beta.name = "tool_beta"

  toolset = skill_toolset.SkillToolset(
      [skill_a, skill_b],
      additional_tools=[tool_alpha, tool_beta],
      max_activated_skills=1,
  )

  # Simulate: skill_a was activated, then skill_b was activated (evicting A)
  readonly_context = mock.create_autospec(ReadonlyContext, instance=True)
  readonly_context.agent_name = "test_agent"
  readonly_context.invocation_id = "inv-1"
  # After eviction, only skill_b is in state
  readonly_context.state.get.return_value = ["skill_b"]

  tools = await toolset.get_tools(readonly_context=readonly_context)
  tool_names = {t.name for t in tools}

  # tool_beta should be present (skill_b is active)
  assert "tool_beta" in tool_names
  # tool_alpha should NOT be present (skill_a was evicted)
  assert "tool_alpha" not in tool_names


@pytest.mark.asyncio
async def test_evicted_tool_call_raises_value_error():
  """Simulates the LLM calling a tool from an evicted skill.

  This mirrors what happens in the real flow: after skill eviction, the
  tool is removed from tools_dict, so _get_tool raises ValueError.
  This test validates that the tools_dict correctly excludes evicted tools.
  """
  # Skill with additional tool
  skill_a = mock.create_autospec(models.Skill, instance=True)
  skill_a.name = "skill_a"
  skill_a.instructions = "instructions A"
  frontmatter_a = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter_a.name = "skill_a"
  frontmatter_a.metadata = {"adk_additional_tools": ["expensive_api_tool"]}
  frontmatter_a.model_dump.return_value = {"name": "skill_a"}
  skill_a.frontmatter = frontmatter_a

  skill_b = mock.create_autospec(models.Skill, instance=True)
  skill_b.name = "skill_b"
  skill_b.instructions = "instructions B"
  frontmatter_b = mock.create_autospec(models.Frontmatter, instance=True)
  frontmatter_b.name = "skill_b"
  frontmatter_b.metadata = {}
  frontmatter_b.model_dump.return_value = {"name": "skill_b"}
  skill_b.frontmatter = frontmatter_b

  expensive_tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
  expensive_tool.name = "expensive_api_tool"

  toolset = skill_toolset.SkillToolset(
      [skill_a, skill_b],
      additional_tools=[expensive_tool],
      max_activated_skills=1,
  )

  # After eviction: only skill_b remains
  readonly_context = mock.create_autospec(ReadonlyContext, instance=True)
  readonly_context.agent_name = "test_agent"
  readonly_context.invocation_id = "inv-2"
  readonly_context.state.get.return_value = ["skill_b"]

  tools = await toolset.get_tools(readonly_context=readonly_context)
  tools_dict = {t.name: t for t in tools}

  # The LLM might still try to call "expensive_api_tool" from history
  assert "expensive_api_tool" not in tools_dict

  # Simulate what _get_tool in functions.py would do
  from google.genai import types as genai_types

  fake_function_call = mock.MagicMock()
  fake_function_call.name = "expensive_api_tool"

  # This mirrors the behavior in flows/llm_flows/functions.py:_get_tool
  with pytest.raises(ValueError, match="Tool 'expensive_api_tool' not found"):
    from google.adk.flows.llm_flows.functions import _get_tool

    _get_tool(fake_function_call, tools_dict)


@pytest.mark.asyncio
async def test_rolling_window_full_lifecycle_with_tool_resolution():
  """End-to-end: load skills, evict, verify tool availability at each step."""
  skills = []
  additional_tools = []
  for i in range(4):
    skill = mock.create_autospec(models.Skill, instance=True)
    skill.name = f"skill_{i}"
    skill.instructions = f"instructions {i}"
    frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
    frontmatter.name = f"skill_{i}"
    frontmatter.metadata = {"adk_additional_tools": [f"tool_{i}"]}
    frontmatter.model_dump.return_value = {"name": f"skill_{i}"}
    skill.frontmatter = frontmatter
    skills.append(skill)

    tool = mock.create_autospec(skill_toolset.BaseTool, instance=True)
    tool.name = f"tool_{i}"
    additional_tools.append(tool)

  toolset = skill_toolset.SkillToolset(
      skills,
      additional_tools=additional_tools,
      max_activated_skills=2,
  )

  load_tool = skill_toolset.LoadSkillTool(toolset)
  ctx = mock.create_autospec(skill_toolset.ToolContext, instance=True)
  ctx._invocation_context = mock.MagicMock()
  ctx.agent_name = "test_agent"
  ctx.invocation_id = "inv-1"

  readonly_context = mock.create_autospec(ReadonlyContext, instance=True)
  readonly_context.agent_name = "test_agent"
  readonly_context.invocation_id = "inv-1"

  # Step 1: Load skill_0 and skill_1
  current_state = []
  for i in range(2):
    ctx.state.get.return_value = list(current_state)
    await load_tool.run_async(
        args={"skill_name": f"skill_{i}"}, tool_context=ctx
    )
    current_state = ctx.state.__setitem__.call_args[0][1]

  assert current_state == ["skill_0", "skill_1"]

  readonly_context.state.get.return_value = list(current_state)
  tools = await toolset.get_tools(readonly_context=readonly_context)
  tool_names = {t.name for t in tools}
  assert "tool_0" in tool_names
  assert "tool_1" in tool_names
  assert "tool_2" not in tool_names

  # Step 2: Load skill_2 — should evict skill_0
  ctx.state.get.return_value = list(current_state)
  await load_tool.run_async(args={"skill_name": "skill_2"}, tool_context=ctx)
  current_state = ctx.state.__setitem__.call_args[0][1]

  assert current_state == ["skill_1", "skill_2"]

  readonly_context.state.get.return_value = list(current_state)
  tools = await toolset.get_tools(readonly_context=readonly_context)
  tool_names = {t.name for t in tools}
  assert "tool_0" not in tool_names  # evicted!
  assert "tool_1" in tool_names
  assert "tool_2" in tool_names

  # Step 3: Load skill_3 — should evict skill_1
  ctx.state.get.return_value = list(current_state)
  await load_tool.run_async(args={"skill_name": "skill_3"}, tool_context=ctx)
  current_state = ctx.state.__setitem__.call_args[0][1]

  assert current_state == ["skill_2", "skill_3"]

  readonly_context.state.get.return_value = list(current_state)
  tools = await toolset.get_tools(readonly_context=readonly_context)
  tool_names = {t.name for t in tools}
  assert "tool_0" not in tool_names
  assert "tool_1" not in tool_names  # evicted!
  assert "tool_2" in tool_names
  assert "tool_3" in tool_names