1  """Regression tests for issue #17335.
 2  
 3  The ``quiet_mode=True`` fast path in :func:`model_tools.get_tool_definitions`
 4  memoizes results to avoid re-walking the registry on every Gateway call. The
 5  cached object must NOT be aliased into callers' return values \u2014 long-lived
 6  Gateway processes mutate the returned list (``run_agent`` appends memory and
 7  LCM context-engine tool schemas to ``self.tools``), and a shared list would
 8  poison subsequent agent inits with duplicate tool names. Providers that
 9  enforce uniqueness (DeepSeek, Xiaomi MiMo, Moonshot/Kimi) then reject the
10  API call with HTTP 400.
11  
12  These tests pin:
13  - the cache-hit path returns a fresh list (existing #17098 behavior)
14  - the first uncached call also returns a fresh list (the fix)
15  - every call returns a list that is not the cached one, even after mutation
16  """
17  from __future__ import annotations
18  
19  import pytest
20  
21  import model_tools
22  
23  
24  @pytest.fixture(autouse=True)
25  def _clear_cache():
26      """Each test starts with an empty quiet_mode cache."""
27      model_tools._tool_defs_cache.clear()
28      yield
29      model_tools._tool_defs_cache.clear()
30  
31  
32  class TestQuietModeCacheIsolation:
33  
34      def test_first_uncached_call_returns_fresh_list(self):
35          """The first quiet_mode call must not alias the cached object \u2014
36          otherwise a caller mutating the returned list mutates the cache."""
37          first = model_tools.get_tool_definitions(quiet_mode=True)
38          assert isinstance(first, list)
39          # Find the cached value to compare identity.
40          assert len(model_tools._tool_defs_cache) == 1
41          cached = next(iter(model_tools._tool_defs_cache.values()))
42          assert first is not cached, (
43              "issue #17335: first quiet_mode call returned the cached list "
44              "by reference \u2014 mutations will leak into subsequent calls."
45          )
46  
47      def test_cache_hit_returns_fresh_list(self):
48          """The cache-hit path already returned a copy pre-fix; pin it."""
49          first = model_tools.get_tool_definitions(quiet_mode=True)
50          second = model_tools.get_tool_definitions(quiet_mode=True)
51          assert first is not second
52          cached = next(iter(model_tools._tool_defs_cache.values()))
53          assert second is not cached
54  
55      def test_caller_mutation_does_not_poison_cache(self):
56          """Simulate run_agent appending LCM tool schemas to the returned
57          list. A second call must NOT see those appended entries."""
58          first = model_tools.get_tool_definitions(quiet_mode=True)
59          baseline_len = len(first)
60          # Caller mutates the returned list (this is what run_agent does
61          # when it injects memory + context-engine tool schemas).
62          first.append({"type": "function", "function": {"name": "lcm_grep"}})
63          first.append({"type": "function", "function": {"name": "lcm_expand"}})
64  
65          second = model_tools.get_tool_definitions(quiet_mode=True)
66          # Length must match the original \u2014 cache pollution would make
67          # second 2 entries longer.
68          assert len(second) == baseline_len, (
69              f"issue #17335: cache was polluted by caller mutation. "
70              f"first len={baseline_len}, mutated len={len(first)}, "
71              f"second-call len={len(second)} \u2014 expected {baseline_len}."
72          )
73          names = [t.get("function", {}).get("name") for t in second]
74          assert "lcm_grep" not in names
75          assert "lcm_expand" not in names
76  
77      def test_repeated_caller_mutation_does_not_accumulate(self):
78          """The original Gateway symptom: every agent init in a long-lived
79          process appends LCM schemas, accumulating duplicates over time."""
80          baseline = len(model_tools.get_tool_definitions(quiet_mode=True))
81          for _ in range(5):
82              tools = model_tools.get_tool_definitions(quiet_mode=True)
83              tools.append({"type": "function", "function": {"name": "lcm_grep"}})
84          final = model_tools.get_tool_definitions(quiet_mode=True)
85          assert len(final) == baseline, (
86              f"Cache accumulated mutations across {5} agent inits: "
87              f"baseline={baseline}, final={len(final)}."
88          )
89  
90      def test_non_quiet_mode_does_not_use_cache(self):
91          """Sanity: quiet_mode=False (TUI path) skips the cache entirely \u2014
92          explains why the bug only hit Gateway."""
93          model_tools.get_tool_definitions(quiet_mode=False)
94          assert len(model_tools._tool_defs_cache) == 0