| 1 | from types import SimpleNamespace |
| 2 | |
| 3 | from extensions.python.message_loop_result import _20_empty_response as empty_response |
| 4 | from extensions.python.message_loop_result._20_empty_response import EmptyResponse |
| 5 | from extensions.python.message_loop_result._30_repeat_response import RepeatResponse |
| 6 | from extensions.python._functions.agent.Agent.hist_add_warning.end import ( |
| 7 | _90_stop_unusable_response_loop as response_loop, |
| 8 | ) |
| 9 | |
| 10 | |
| 11 | class FakeAgent: |
| 12 | def __init__(self, response: str, reasoning: str = "", last_response: str = ""): |
| 13 | self.loop_data = SimpleNamespace( |
| 14 | last_response=last_response, |
| 15 | params_temporary={}, |
| 16 | params_persistent={}, |
| 17 | iteration=0, |
| 18 | ) |
| 19 | self.logs = [] |
| 20 | self.context = SimpleNamespace( |
| 21 | log=SimpleNamespace(log=lambda **entry: self.logs.append(entry)) |
| 22 | ) |
| 23 | self.agent_name = "A0" |
| 24 | self.response = response |
| 25 | self.reasoning = reasoning |
| 26 | self.warnings = [] |
| 27 | self.history = [] |
| 28 | |
| 29 | def read_prompt(self, name, **kwargs): |
| 30 | if name == "fw.msg_unusable_response_limit.md": |
| 31 | return f"stopped at {kwargs['limit']}" |
| 32 | return { |
| 33 | "fw.msg_misformat.md": "misformatted", |
| 34 | "fw.msg_empty_response.md": "empty", |
| 35 | "fw.msg_repeat.md": "repeat", |
| 36 | "fw.msg_repeat_response.md": "Repeated response detected. Retrying.", |
| 37 | }[name] |
| 38 | |
| 39 | def hist_add_ai_response(self, response, **kwargs): |
| 40 | self.history.append(response) |
| 41 | return SimpleNamespace(id="assistant") |
| 42 | |
| 43 | def _remember_llm_result_state(self, *args): |
| 44 | pass |
| 45 | |
| 46 | def hist_add_warning(self, message): |
| 47 | self.warnings.append(message) |
| 48 | return SimpleNamespace(id="warning") |
| 49 | |
| 50 | |
| 51 | def _run(agent): |
| 52 | result_data = { |
| 53 | "llm_result": SimpleNamespace(response=agent.response, reasoning=agent.reasoning) |
| 54 | } |
| 55 | EmptyResponse(agent).execute(result_data) |
| 56 | RepeatResponse(agent).execute(result_data) |
| 57 | return result_data |
| 58 | |
| 59 | |
| 60 | def test_empty_result_skips_default_processing(): |
| 61 | agent = FakeAgent("") |
| 62 | |
| 63 | assert _run(agent)["skip_default_processing"] is True |
| 64 | assert agent.history == [] |
| 65 | assert agent.warnings == [] |
| 66 | assert agent.logs == [{"type": "warning", "content": "A0: empty"}] |
| 67 | |
| 68 | |
| 69 | def test_empty_result_counts_toward_unusable_response_limit(monkeypatch): |
| 70 | monkeypatch.setattr( |
| 71 | empty_response, |
| 72 | "get_settings", |
| 73 | lambda: {"max_consecutive_unusable_responses": 2}, |
| 74 | ) |
| 75 | agent = FakeAgent("") |
| 76 | |
| 77 | assert _run(agent)["skip_default_processing"] is True |
| 78 | |
| 79 | agent.loop_data.iteration = 1 |
| 80 | try: |
| 81 | _run(agent) |
| 82 | except response_loop.HandledException as error: |
| 83 | assert str(error) == "stopped at 2" |
| 84 | else: |
| 85 | raise AssertionError("empty response should stop at the configured limit") |
| 86 | |
| 87 | assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 2 |
| 88 | |
| 89 | |
| 90 | def test_later_handlers_skip_a_result_already_handled_by_an_extension(): |
| 91 | response = '{"tool_name":"response"}' |
| 92 | agent = FakeAgent(response, last_response=response) |
| 93 | result_data = { |
| 94 | "llm_result": SimpleNamespace(response=response, reasoning=""), |
| 95 | "skip_default_processing": True, |
| 96 | } |
| 97 | |
| 98 | EmptyResponse(agent).execute(result_data) |
| 99 | RepeatResponse(agent).execute(result_data) |
| 100 | |
| 101 | assert agent.history == [] |
| 102 | assert agent.warnings == [] |
| 103 | |
| 104 | |
| 105 | def test_repeat_skips_default_processing(): |
| 106 | agent = FakeAgent('{"tool_name":"response"}', last_response='{"tool_name":"response"}') |
| 107 | |
| 108 | assert _run(agent)["skip_default_processing"] is True |
| 109 | assert agent.warnings == ["repeat"] |
| 110 | assert agent.logs == [ |
| 111 | { |
| 112 | "type": "warning", |
| 113 | "content": "A0: Repeated response detected. Retrying.", |
| 114 | "id": "warning", |
| 115 | } |
| 116 | ] |
| 117 | |
| 118 | |
| 119 | def test_repeat_ignores_reasoning(): |
| 120 | response = '{"tool_name":"response"}' |
| 121 | agent = FakeAgent(response, reasoning="thinking", last_response=response) |
| 122 | |
| 123 | assert _run(agent)["skip_default_processing"] is True |
| 124 | assert agent.warnings == ["repeat"] |
| 125 | |
| 126 | |
| 127 | def test_result_with_reasoning_uses_default_processing(): |
| 128 | agent = FakeAgent("", reasoning="thinking", last_response="previous") |
| 129 | |
| 130 | assert "skip_default_processing" not in _run(agent) |
| 131 | assert agent.history == [] |