| 1 | from pathlib import Path |
| 2 | from types import SimpleNamespace |
| 3 | |
| 4 | from helpers.errors import HandledException |
| 5 | from helpers.files import read_prompt_file |
| 6 | from helpers.settings import get_default_settings, normalize_settings |
| 7 | from extensions.python._functions.agent.Agent.hist_add_warning.end import ( |
| 8 | _90_stop_unusable_response_loop as response_loop, |
| 9 | ) |
| 10 | |
| 11 | |
| 12 | class FakeLog: |
| 13 | def __init__(self): |
| 14 | self.entries = [] |
| 15 | |
| 16 | def log(self, **entry): |
| 17 | self.entries.append(entry) |
| 18 | |
| 19 | |
| 20 | def _agent(): |
| 21 | prompts = { |
| 22 | "fw.msg_misformat.md": "misformatted", |
| 23 | "fw.msg_repeat.md": "repeated", |
| 24 | "fw.msg_empty_response.md": "empty response", |
| 25 | } |
| 26 | |
| 27 | def read_prompt(name, **kwargs): |
| 28 | if name == "fw.msg_unusable_response_limit.md": |
| 29 | return f"stopped at {kwargs['limit']}" |
| 30 | return prompts[name] |
| 31 | |
| 32 | return SimpleNamespace( |
| 33 | loop_data=SimpleNamespace(iteration=0, params_persistent={}), |
| 34 | context=SimpleNamespace(log=FakeLog()), |
| 35 | read_prompt=read_prompt, |
| 36 | ) |
| 37 | |
| 38 | |
| 39 | def _run(extension, agent, message): |
| 40 | data = {"args": (agent, message), "kwargs": {}, "exception": None} |
| 41 | extension.execute(data=data) |
| 42 | return data |
| 43 | |
| 44 | |
| 45 | def test_stops_at_configured_failure_limit(monkeypatch): |
| 46 | monkeypatch.setattr( |
| 47 | response_loop, |
| 48 | "get_settings", |
| 49 | lambda: {"max_consecutive_unusable_responses": 3}, |
| 50 | ) |
| 51 | agent = _agent() |
| 52 | extension = response_loop.StopUnusableResponseLoop(agent=agent) |
| 53 | |
| 54 | assert _run(extension, agent, "misformatted")["exception"] is None |
| 55 | |
| 56 | agent.loop_data.iteration = 1 |
| 57 | assert _run(extension, agent, "empty response")["exception"] is None |
| 58 | |
| 59 | agent.loop_data.iteration = 2 |
| 60 | data = _run(extension, agent, "repeated") |
| 61 | |
| 62 | assert isinstance(data["exception"], HandledException) |
| 63 | assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 3 |
| 64 | assert agent.context.log.entries == [ |
| 65 | {"type": "warning", "content": "stopped at 3"} |
| 66 | ] |
| 67 | |
| 68 | |
| 69 | def test_nonconsecutive_failure_starts_a_new_recovery_window(monkeypatch): |
| 70 | monkeypatch.setattr( |
| 71 | response_loop, |
| 72 | "get_settings", |
| 73 | lambda: {"max_consecutive_unusable_responses": 2}, |
| 74 | ) |
| 75 | agent = _agent() |
| 76 | extension = response_loop.StopUnusableResponseLoop(agent=agent) |
| 77 | |
| 78 | assert _run(extension, agent, {"structured": "warning"})["exception"] is None |
| 79 | _run(extension, agent, "misformatted") |
| 80 | agent.loop_data.iteration = 2 |
| 81 | data = _run(extension, agent, "repeated") |
| 82 | |
| 83 | assert data["exception"] is None |
| 84 | assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 1 |
| 85 | |
| 86 | |
| 87 | def test_general_settings_expose_the_default_failure_limit(): |
| 88 | settings = get_default_settings() |
| 89 | assert settings["max_consecutive_unusable_responses"] == 5 |
| 90 | settings["max_consecutive_unusable_responses"] = 0 |
| 91 | assert normalize_settings(settings)["max_consecutive_unusable_responses"] == 1 |
| 92 | |
| 93 | html = Path("webui/components/settings/agent/agent.html").read_text() |
| 94 | assert ( |
| 95 | 'x-model.number="$store.settings.settings.max_consecutive_unusable_responses"' |
| 96 | in html |
| 97 | ) |
| 98 | assert "after 3 consecutive" in read_prompt_file( |
| 99 | "fw.msg_unusable_response_limit.md", ["prompts"], limit=3 |
| 100 | ) |