From ff09b98d2f0bc945caeba740c44a67a846fb7e6c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:38:06 +0000 Subject: [PATCH] test: unit tests for dynamic_prompting, generic_agent and reproducibility_agent (feature: test coverage; xfail bug markers) Co-Authored-By: Daniella Grimberg --- tests/agents/test_dynamic_prompting_unit.py | 586 ++++++++++++++++++ tests/agents/test_generic_agent_unit.py | 517 +++++++++++++++ .../agents/test_reproducibility_agent_unit.py | 357 +++++++++++ 3 files changed, 1460 insertions(+) create mode 100644 tests/agents/test_dynamic_prompting_unit.py create mode 100644 tests/agents/test_generic_agent_unit.py create mode 100644 tests/agents/test_reproducibility_agent_unit.py diff --git a/tests/agents/test_dynamic_prompting_unit.py b/tests/agents/test_dynamic_prompting_unit.py new file mode 100644 index 000000000..833badc21 --- /dev/null +++ b/tests/agents/test_dynamic_prompting_unit.py @@ -0,0 +1,586 @@ +"""Unit tests for agentlab.agents.dynamic_prompting (no browser, no tokenizer download).""" + +import logging +from unittest import mock + +import numpy as np +import pytest +from bgym import HighLevelActionSetArgs + +from agentlab.agents import dynamic_prompting as dp +from agentlab.llm.llm_utils import HumanMessage, ParseError + + +def _word_count(text, model=None): + return len(text.split()) + + +@pytest.fixture +def fake_tokens(monkeypatch): + """Replace tiktoken/HF token counting with a deterministic word count.""" + monkeypatch.setattr(dp, "count_tokens", _word_count) + return _word_count + + +def _obs(**overrides): + obs = { + "pruned_html": "
hello
", + "axtree_txt": "[12] button 'Submit'", + "last_action_error": "", + "focused_element_bid": "12", + "open_pages_urls": ["https://a.example", "https://b.example"], + "open_pages_titles": ["A", "B"], + "active_page_index": 1, + "screenshot": np.zeros((4, 4, 3), dtype=np.uint8), + "screenshot_som": np.full((4, 4, 3), 255, dtype=np.uint8), + } + obs.update(overrides) + return obs + + +def _action_set(multiaction=False, strict=False): + return HighLevelActionSetArgs( + subsets=["bid"], multiaction=multiaction, strict=strict + ).make_action_set() + + +# --------------------------------------------------------------------------- Flags + + +class TestFlags: + def test_copy_is_deep_and_independent(self): + flags = dp.ObsFlags(use_html=False) + clone = flags.copy() + assert clone == flags and clone is not flags + clone.use_html = True + assert flags.use_html is False + + def test_asdict_roundtrip_through_from_dict(self): + flags = dp.ObsFlags(use_ax_tree=True, html_type="dom_txt", extract_coords="box") + as_dict = flags.asdict() + assert as_dict["use_ax_tree"] is True + assert as_dict["extract_coords"] == "box" + assert dp.ObsFlags.from_dict(as_dict) == flags + + def test_from_dict_returns_obsflags_instance_unchanged(self): + flags = dp.ObsFlags() + assert dp.ObsFlags.from_dict(flags) is flags + + @pytest.mark.parametrize("bad", [None, 3, "use_html", ["use_html"]]) + def test_from_dict_rejects_non_dict(self, bad): + with pytest.raises(ValueError, match="Unregcognized type"): + dp.ObsFlags.from_dict(bad) + + def test_action_flags_defaults(self): + flags = dp.ActionFlags() + assert flags.action_set is None + assert flags.long_description is True + assert flags.individual_examples is False + assert flags.multi_actions is None and flags.is_strict is None + + @pytest.mark.xfail( + strict=True, + reason="BUG: Flags.from_dict ignores cls and always builds ObsFlags, so ActionFlags.from_dict fails", + ) + def test_action_flags_from_dict_returns_action_flags(self): + flags = dp.ActionFlags(long_description=False) + restored = dp.ActionFlags.from_dict(flags.asdict()) + assert isinstance(restored, dp.ActionFlags) + assert restored == flags + + +# --------------------------------------------------------------------------- PromptElement + + +class _Answering(dp.PromptElement): + _prompt = "P" + _abstract_ex = "A" + _concrete_ex = "C" + + def _parse_answer(self, text_answer): + return {"echo": text_answer} + + +class TestPromptElement: + def test_visible_exposes_all_parts(self): + el = _Answering() + assert (el.prompt, el.abstract_ex, el.concrete_ex) == ("P", "A", "C") + assert el.parse_answer("x") == {"echo": "x"} + + def test_hidden_returns_empty_everything(self): + el = _Answering(visible=False) + assert (el.prompt, el.abstract_ex, el.concrete_ex) == ("", "", "") + assert el.parse_answer("x") == {} + + def test_callable_visibility_is_reevaluated(self): + state = {"on": False} + el = _Answering(visible=lambda: state["on"]) + assert el.prompt == "" + state["on"] = True + assert el.prompt == "P" + + def test_base_parse_answer_returns_empty_dict(self): + assert dp.PromptElement().parse_answer("x") == {} + + +# --------------------------------------------------------------------------- Trunkater / fit_tokens + + +class _Lines(dp.Trunkater): + def __init__(self, n_lines, visible=True, start=0, speed=0.5): + super().__init__(visible=visible, shrink_speed=speed, start_trunkate_iteration=start) + self._prompt = "\n".join(f"line {i}" for i in range(n_lines)) + + +class TestTrunkater: + def test_no_truncation_before_start_iteration(self): + el = _Lines(10, start=2) + original = el.prompt + el.shrink() + el.shrink() + assert el.prompt == original + assert el.shrink_calls == 2 + el.shrink() + assert el.prompt != original + + def test_truncation_removes_fraction_and_notes_deleted_lines(self): + el = _Lines(10, start=0, speed=0.5) + el.shrink() + lines = el.prompt.splitlines() + assert lines[:5] == [f"line {i}" for i in range(5)] + assert lines[-1] == "... Deleted 5 lines to reduce prompt size." + assert el.deleted_lines == 5 + el.shrink() + assert el.deleted_lines == 5 + (6 - 3) + assert el.prompt.endswith(f"Deleted {el.deleted_lines} lines to reduce prompt size.") + + def test_hidden_trunkater_does_not_truncate_but_counts_calls(self): + el = _Lines(10, visible=False, start=0) + raw = el._prompt + el.shrink() + assert el._prompt == raw + assert el.shrink_calls == 1 + + +class _Shrinker(dp.Shrinkable): + def __init__(self, prompts): + super().__init__() + self.prompts = list(prompts) + self.shrinks = 0 + + @property + def _prompt(self): + return self.prompts[min(self.shrinks, len(self.prompts) - 1)] + + def shrink(self): + self.shrinks += 1 + + +class TestFitTokens: + def test_no_max_returns_prompt_without_counting(self, monkeypatch): + monkeypatch.setattr( + dp, "count_tokens", mock.Mock(side_effect=AssertionError("should not count")) + ) + el = _Shrinker(["a b c"]) + assert dp.fit_tokens(el, max_prompt_tokens=None) == "a b c" + assert el.shrinks == 0 + + def test_shrinks_until_fits(self, fake_tokens): + el = _Shrinker(["w " * 50, "w " * 20, "w " * 5]) + out = dp.fit_tokens(el, max_prompt_tokens=10, additional_prompts=[]) + assert out == "w " * 5 + assert el.shrinks == 2 + + def test_additional_prompt_string_reduces_budget(self, fake_tokens): + # budget 10 - (3 words + 1) = 6 -> "w "*5 fits, "w "*8 does not + el = _Shrinker(["w " * 8, "w " * 5]) + out = dp.fit_tokens(el, max_prompt_tokens=10, additional_prompts="x y z") + assert out == "w " * 5 + assert el.shrinks == 1 + + def test_additional_prompt_list_reduces_budget(self, fake_tokens): + el = _Shrinker(["w " * 8, "w " * 5]) + out = dp.fit_tokens(el, max_prompt_tokens=10, additional_prompts=["x", "y z"]) + # budget 10 - 2 - 3 = 5 + assert out == "w " * 5 + + def test_returns_last_prompt_and_logs_after_max_iterations(self, fake_tokens, caplog): + el = _Shrinker(["w " * 100]) + with caplog.at_level(logging.INFO): + out = dp.fit_tokens(el, max_prompt_tokens=3, max_iterations=4, additional_prompts=[]) + assert out == "w " * 100 + assert el.shrinks == 4 + assert "After 4 shrink iterations" in caplog.text + + def test_list_prompt_is_deprecated_but_counted(self, fake_tokens): + prompt = [ + {"type": "text", "text": "one two"}, + {"type": "image_url", "image_url": "data:xxx"}, + ] + el = _Shrinker([prompt]) + with pytest.warns(DeprecationWarning, match="list of prompts is deprecated"): + out = dp.fit_tokens(el, max_prompt_tokens=5, additional_prompts=[]) + assert out is prompt + assert el.shrinks == 0 + + def test_base_message_prompt_is_counted_without_images(self, fake_tokens): + msg = HumanMessage("one two three") + msg.add_image("data:image/png;base64,AAAA") + el = _Shrinker([msg]) + assert dp.fit_tokens(el, max_prompt_tokens=4, additional_prompts=[]) is msg + + def test_unknown_prompt_type_raises(self, fake_tokens): + el = _Shrinker([12345]) + with pytest.raises(ValueError, match="Unrecognized type for prompt"): + dp.fit_tokens(el, max_prompt_tokens=10, additional_prompts=[]) + + +# --------------------------------------------------------------------------- Observation parts + + +class TestHTMLAndAXTree: + def test_html_with_and_without_visible_note(self): + with_note = dp.HTML("", visible_elements_only=True, prefix="## ") + assert with_note.prompt.startswith("\n## HTML:\nNote: only elements that are visible") + assert with_note.prompt.endswith("\n") + assert "Note:" not in dp.HTML("", visible_elements_only=False).prompt + assert with_note.start_trunkate_iteration == 5 + + @pytest.mark.parametrize( + "coord_type, expected, absent", + [ + ("center", "center coordinates are provided", "bounding box"), + ("box", "bounding box of each object", "center coordinates"), + (None, None, "coordinates are provided"), + ("False", None, "bounding box"), + ], + ) + def test_axtree_coord_notes(self, coord_type, expected, absent): + prompt = dp.AXTree("[1] x", visible_elements_only=False, coord_type=coord_type).prompt + if expected: + assert expected in prompt + assert absent not in prompt + assert "[bid] is the unique alpha-numeric identifier" in prompt + + def test_axtree_visible_tag_and_visible_only_notes(self): + p = dp.AXTree("[1] x", visible_elements_only=True, visible_tag=True, prefix="## ").prompt + assert p.startswith("\n## AXTree:") + assert "only elements that are visible in the viewport" in p + assert 'If the "visible" tag is not' in p + p2 = dp.AXTree("[1] x", visible_elements_only=False, visible_tag=False).prompt + assert "visible in the viewport" not in p2 + assert '"visible" tag' not in p2 + + +class TestError: + def test_plain_error(self): + assert dp.Error("boom", prefix="## ").prompt == "\n## Error from previous action:\nboom\n" + + def test_call_log_truncated_to_10_lines(self): + logs = "\n".join(f"log{i}" for i in range(30)) + prompt = dp.Error(f"Timeout\nCall log:{logs}").prompt + assert "Call log:" in prompt + assert "log9" in prompt + assert "log10" not in prompt and "log29" not in prompt + + def test_call_log_kept_when_limit_disabled(self): + logs = "\n".join(f"log{i}" for i in range(30)) + prompt = dp.Error(f"Timeout\nCall log:{logs}", limit_logs=False).prompt + assert "log29" in prompt + + +class TestFocusedElementAndTabs: + @pytest.mark.parametrize("bid, expected", [("a12", "bid='a12'"), ("", "None"), (None, "None")]) + def test_focused_element(self, bid, expected): + prompt = dp.FocusedElement(bid, prefix="## ").prompt + assert prompt.startswith("\n## Focused element:\n") + assert prompt.strip().endswith(expected) + + def test_tabs_marks_active_tab(self): + prompt = dp.Tabs(_obs(), prefix="## ").prompt + assert "## Currently open tabs:" in prompt + assert "Tab 0:\n Title: A\n URL: https://a.example" in prompt + assert "Tab 1 (active tab):\n Title: B" in prompt + + def test_hidden_tabs_do_not_touch_obs(self): + assert dp.Tabs({}, visible=False).prompt == "" + + +class TestObservation: + def test_prompt_respects_flags(self): + flags = dp.ObsFlags( + use_html=False, + use_ax_tree=True, + use_tabs=True, + use_focused_element=True, + use_error_logs=True, + ) + prompt = dp.Observation(_obs(last_action_error="kaboom"), flags).prompt + assert prompt.startswith("\n# Observation of current step:") + assert "## AXTree:" in prompt + assert "## HTML:" not in prompt + assert "Currently open tabs" in prompt + assert "bid='12'" in prompt + assert "kaboom" in prompt + + def test_error_hidden_when_no_error(self): + flags = dp.ObsFlags(use_error_logs=True) + assert "Error from previous action" not in dp.Observation(_obs(), flags).prompt + + def test_shrink_truncates_html_after_its_start_iteration(self): + html = "\n".join(f"

{i}

" for i in range(100)) + obs = dp.Observation(_obs(pruned_html=html), dp.ObsFlags(use_html=True)) + before = obs.prompt + for _ in range(5): + obs.shrink() + assert obs.prompt == before + obs.shrink() + assert "Deleted" in obs.prompt and len(obs.prompt) < len(before) + assert obs.ax_tree.shrink_calls == 6 + + @pytest.mark.parametrize( + "use_som, text, key", + [ + (False, "screenshot of the page:", "screenshot"), + (True, "bounding boxes", "screenshot_som"), + ], + ) + def test_add_screenshot(self, use_som, text, key, monkeypatch): + seen = {} + + def fake_b64(img): + seen["img"] = img + return "data:image/jpeg;base64,XYZ" + + monkeypatch.setattr(dp, "image_to_jpg_base64_url", fake_b64) + obs_dict = _obs() + flags = dp.ObsFlags(use_screenshot=True, use_som=use_som, openai_vision_detail="low") + msg = dp.Observation(obs_dict, flags).add_screenshot(HumanMessage("hi")) + assert seen["img"] is obs_dict[key] + assert text in msg["content"][1]["text"] + assert msg["content"][2] == { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,XYZ", "detail": "low"}, + } + + def test_add_screenshot_disabled_is_noop(self): + msg = HumanMessage("hi") + out = dp.Observation(_obs(), dp.ObsFlags(use_screenshot=False)).add_screenshot(msg) + assert out is msg and out["content"] == "hi" + + def test_add_screenshot_real_encoding(self): + flags = dp.ObsFlags(use_screenshot=True, openai_vision_detail="auto") + msg = dp.Observation(_obs(), flags).add_screenshot(HumanMessage("hi")) + assert msg["content"][-1]["image_url"]["url"].startswith("data:image/jpeg;base64,") + + +# --------------------------------------------------------------------------- Misc elements + + +class TestMiscElements: + @pytest.mark.parametrize("system, visible", [("Darwin", True), ("Linux", False)]) + def test_mac_note(self, monkeypatch, system, visible): + monkeypatch.setattr(dp.platform, "system", lambda: system) + assert bool(dp.MacNote().prompt) is visible + + def test_be_cautious(self): + assert "Be very cautious" in dp.BeCautious().prompt + assert dp.BeCautious(visible=False).prompt == "" + + def test_goal_instructions_with_extra(self): + goal = [{"type": "text", "text": "Buy milk"}] + el = dp.GoalInstructions(goal, extra_instructions="Be fast") + prompt = el.prompt + assert prompt[0]["text"].startswith("# Instructions") + assert prompt[1] == {"type": "text", "text": "Buy milk"} + assert "## Extra instructions:" in prompt[2]["text"] + assert "Be fast" in prompt[2]["text"] + assert len(dp.GoalInstructions(goal).prompt) == 2 + + def test_chat_instructions_lists_messages(self, monkeypatch): + monkeypatch.setenv("TZ", "UTC") + msgs = [ + {"role": "user", "timestamp": 0, "message": "hello"}, + {"role": "assistant", "timestamp": 60, "message": "hi there"}, + ] + prompt = dp.ChatInstructions(msgs, extra_instructions="be nice").prompt + assert "## Chat messages:" in prompt + assert "- [user] UTC Time: Thu Jan 1 00:00:00 1970" in prompt + assert "- [assistant] UTC Time: Thu Jan 1 00:01:00 1970" in prompt + assert prompt.rstrip().endswith("be nice") + assert "Extra instructions" not in dp.ChatInstructions(msgs).prompt + + def test_static_prompts(self): + assert "bid to identify elements" in dp.Hints().prompt + assert dp.SystemPrompt().prompt.startswith("You are an agent") + + +# --------------------------------------------------------------------------- ActionPrompt + + +class TestActionPrompt: + def test_prompt_and_examples(self, monkeypatch): + monkeypatch.setattr(dp.platform, "system", lambda: "Linux") + ap = dp.ActionPrompt(_action_set(), dp.ActionFlags(long_description=False)) + assert ap.prompt.startswith("# Action space:\nNote: This action set allows") + assert "click(bid" in ap.prompt + assert "Description:" not in ap.prompt + assert "One single action" in ap.abstract_ex + assert "click('a324')" in ap.concrete_ex + + def test_parse_valid_action(self): + ap = dp.ActionPrompt(_action_set(), dp.ActionFlags()) + assert ap.parse_answer("click('12')") == {"action": "click('12')"} + + def test_parse_none_action_maps_to_python_none(self): + ap = dp.ActionPrompt(_action_set(), dp.ActionFlags()) + assert ap.parse_answer("None") == {"action": None} + + def test_missing_tag_strict_raises(self): + ap = dp.ActionPrompt(_action_set(), dp.ActionFlags(is_strict=True)) + with pytest.raises(ParseError, match="Missing the key "): + ap.parse_answer("```\nclick('12')\n```") + + def test_missing_tag_non_strict_falls_back_to_code_blocks(self): + ap = dp.ActionPrompt(_action_set(multiaction=True), dp.ActionFlags(is_strict=False)) + ans = ap.parse_answer("```python\nclick('1')\n```\ntext\n```\nclick('2')\n```") + assert ans["action"] == "click('1')\nclick('2')" + assert "Missing the key " in ans["parse_error"] + + def test_missing_tag_non_strict_without_code_blocks_raises(self): + ap = dp.ActionPrompt(_action_set(), dp.ActionFlags(is_strict=False)) + with pytest.raises(ParseError, match="Missing the key "): + ap.parse_answer("I would click the button") + + @pytest.mark.parametrize("bad", ["", "click('1')\nclick('2')"]) + def test_unmappable_action_raises_parse_error(self, bad): + ap = dp.ActionPrompt(_action_set(multiaction=False), dp.ActionFlags()) + with pytest.raises(ParseError, match="Error while parsing action"): + ap.parse_answer(f"{bad}") + + +class TestThink: + def test_parses_think(self): + assert dp.Think().parse_answer("hmmx") == {"think": "hmm"} + + def test_missing_tag_returns_whole_text_with_error(self): + ans = dp.Think().parse_answer("just text") + assert ans["think"] == "just text" + assert "Missing the key " in ans["parse_error"] + + def test_examples(self): + assert "" in dp.Think().abstract_ex and "" in dp.Think().concrete_ex + assert dp.Think(visible=False).parse_answer("x") == {} + + +# --------------------------------------------------------------------------- History + + +class TestHistory: + def _history(self, flags, memories=("m0", "m1")): + obs = [ + _obs(last_action_error=""), + _obs(last_action_error="err at 1"), + _obs(last_action_error=""), + ] + return dp.History( + obs, + ["click('1')", "click('2')"], + list(memories) if memories is not None else None, + ["t0", "t1"], + flags, + ) + + def test_full_history_prompt(self): + flags = dp.ObsFlags( + use_history=True, + use_think_history=True, + use_action_history=True, + use_error_logs=True, + use_past_error_logs=True, + ) + prompt = self._history(flags).prompt + assert prompt.startswith("# History of interaction with the task:\n") + assert "## step 0" in prompt and "## step 1" in prompt + assert "\nt0\n" in prompt + assert "\nclick('2')\n" in prompt + assert "\nm1\n" in prompt + assert prompt.count("### Error from previous action") == 1 + assert "err at 1" in prompt + + def test_history_hidden_and_sparse_flags(self): + assert self._history(dp.ObsFlags(use_history=False)).prompt == "" + prompt = self._history(dp.ObsFlags(use_history=True), memories=None).prompt + assert "" not in prompt and "" not in prompt + assert "" not in prompt and "Error" not in prompt + + def test_past_errors_hidden_without_past_error_flag(self): + flags = dp.ObsFlags(use_history=True, use_error_logs=True, use_past_error_logs=False) + assert "err at 1" not in self._history(flags).prompt + + def test_length_mismatch_asserts(self): + with pytest.raises(AssertionError): + dp.History([_obs()], ["click('1')"], None, ["t"], dp.ObsFlags()) + + def test_shrink_propagates_to_steps(self): + h = self._history(dp.ObsFlags(use_history=True)) + with mock.patch.object(dp.HistoryStep, "shrink", autospec=True) as step_shrink: + h.shrink() + assert step_shrink.call_count == 2 + # real shrink on HistoryStep is a no-op that must not raise + h.history_steps[0].shrink() + + +# --------------------------------------------------------------------------- obs preprocessor + + +def test_make_obs_preprocessor_passes_flags_and_builds_fields(monkeypatch): + calls = {} + + def fake_flatten(name): + def _f(obj, **kwargs): + calls[name] = (obj, kwargs) + return f"{name}:{obj}" + + return _f + + monkeypatch.setattr(dp, "flatten_dom_to_str", fake_flatten("dom")) + monkeypatch.setattr(dp, "flatten_axtree_to_str", fake_flatten("ax")) + monkeypatch.setattr(dp, "prune_html", lambda s: f"pruned({s})") + monkeypatch.setattr( + dp, "overlay_som", lambda img, extra_properties: ("som", img, extra_properties) + ) + + flags = dp.ObsFlags( + extract_visible_tag=True, + extract_clickable_tag=False, + extract_coords="center", + filter_visible_elements_only=True, + filter_with_bid_only=True, + filter_som_only=False, + ) + raw = { + "dom_object": "D", + "axtree_object": "A", + "extra_element_properties": {"x": 1}, + "screenshot": "IMG", + } + out = dp.make_obs_preprocessor(flags)(raw) + + assert "dom_txt" not in raw # input is not mutated + assert out["dom_txt"] == "dom:D" + assert out["axtree_txt"] == "ax:A" + assert out["pruned_html"] == "pruned(dom:D)" + assert out["screenshot_som"] == ("som", "IMG", {"x": 1}) + for name in ("dom", "ax"): + kwargs = calls[name][1] + assert kwargs == dict( + extra_properties={"x": 1}, + with_visible=True, + with_clickable=False, + with_center_coords=True, + with_bounding_box_coords=False, + filter_visible_only=True, + filter_with_bid_only=True, + filter_som_only=False, + ) diff --git a/tests/agents/test_generic_agent_unit.py b/tests/agents/test_generic_agent_unit.py new file mode 100644 index 000000000..7ace9204b --- /dev/null +++ b/tests/agents/test_generic_agent_unit.py @@ -0,0 +1,517 @@ +"""Unit tests for GenericAgent / GenericAgentArgs / generic_agent_prompt with a fake LLM.""" + +import logging +from dataclasses import dataclass, field +from types import SimpleNamespace +from unittest import mock + +import pytest +from bgym import HighLevelActionSetArgs + +from agentlab.agents import dynamic_prompting as dp +from agentlab.agents.generic_agent import generic_agent as ga +from agentlab.agents.generic_agent.generic_agent import GenericAgent, GenericAgentArgs +from agentlab.agents.generic_agent.generic_agent_prompt import ( + Criticise, + GenericPromptFlags, + MainPrompt, + Memory, + Plan, +) +from agentlab.llm.base_api import BaseModelArgs +from agentlab.llm.chat_api import make_assistant_message +from agentlab.llm.llm_utils import HumanMessage, ParseError + + +class FakeChat: + """Returns scripted assistant messages and records each prompt it sees.""" + + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def __call__(self, messages): + self.calls.append([dict(m) for m in messages]) + content = self.responses[min(len(self.calls) - 1, len(self.responses) - 1)] + return make_assistant_message(content) + + def get_stats(self): + return {"n_calls": len(self.calls)} + + +@dataclass +class FakeModelArgs(BaseModelArgs): + model_name: str = "fake/model" + responses: tuple = ("okclick('12')",) + prepared: list = field(default_factory=list) + + def make_model(self): + return FakeChat(self.responses) + + def prepare_server(self): + self.prepared.append("prepare") + return "prepared" + + def close_server(self): + self.prepared.append("close") + return "closed" + + +def make_flags(**overrides): + obs_kwargs = overrides.pop("obs", {}) + action_kwargs = overrides.pop("action", {}) + obs = dict( + use_html=True, + use_ax_tree=True, + use_history=True, + use_action_history=True, + use_think_history=True, + use_screenshot=False, + ) + obs.update(obs_kwargs) + action = dict(action_set=HighLevelActionSetArgs(subsets=["bid"])) + action.update(action_kwargs) + kwargs = dict( + obs=dp.ObsFlags(**obs), + action=dp.ActionFlags(**action), + use_thinking=True, + max_prompt_tokens=None, + ) + kwargs.update(overrides) + return GenericPromptFlags(**kwargs) + + +def make_obs(step=0, **overrides): + obs = { + "goal_object": [{"type": "text", "text": "Click the submit button"}], + "chat_messages": [{"role": "user", "message": "Click submit", "timestamp": 0}], + "pruned_html": f"", + "axtree_txt": f"[12] button 'Submit {step}'", + "last_action_error": "", + "focused_element_bid": "", + "open_pages_urls": ["https://example.com"], + "open_pages_titles": ["Example"], + "active_page_index": 0, + } + obs.update(overrides) + return obs + + +def make_agent(responses=None, flags=None, max_retry=4, **model_kwargs): + if responses is not None: + model_kwargs["responses"] = tuple(responses) + args = GenericAgentArgs( + chat_model_args=FakeModelArgs(**model_kwargs), + flags=flags or make_flags(), + max_retry=max_retry, + ) + return args.make_agent() + + +# --------------------------------------------------------------------------- GenericAgentArgs + + +class TestGenericAgentArgs: + def test_agent_name_derived_from_model_and_slashes_replaced(self): + args = GenericAgentArgs(chat_model_args=FakeModelArgs(model_name="org/m-1")) + assert args.agent_name == "GenericAgent-org_m-1" + + def test_missing_model_args_keeps_default_name(self): + assert GenericAgentArgs().agent_name is None + + def _benchmark(self, name="miniwob_tiny", multi_tab=True): + return SimpleNamespace( + name=name, + is_multi_tab=multi_tab, + high_level_action_set_args=HighLevelActionSetArgs( + subsets=["miniwob_all"], multiaction=False, strict=False + ), + ) + + def test_set_benchmark_miniwob_enables_html_and_copies_action_set(self): + flags = make_flags(obs=dict(use_html=False)) + args = GenericAgentArgs(chat_model_args=FakeModelArgs(), flags=flags) + bench = self._benchmark() + args.set_benchmark(bench, demo_mode=False) + assert flags.obs.use_html is True + assert flags.obs.use_tabs is True + assert flags.action.action_set == bench.high_level_action_set_args + assert flags.action.action_set is not bench.high_level_action_set_args + assert flags.action.action_set.demo_mode is None + + def test_set_benchmark_non_miniwob_keeps_html_flag(self): + flags = make_flags(obs=dict(use_html=False)) + args = GenericAgentArgs(chat_model_args=FakeModelArgs(), flags=flags) + args.set_benchmark(self._benchmark(name="workarena_l1", multi_tab=False), demo_mode=False) + assert flags.obs.use_html is False + assert flags.obs.use_tabs is False + + def test_set_benchmark_legacy_flags_and_demo_mode(self): + flags = make_flags(action=dict(multi_actions=True, is_strict=True)) + args = GenericAgentArgs(chat_model_args=FakeModelArgs(), flags=flags) + bench = self._benchmark() + args.set_benchmark(bench, demo_mode=True) + action_set = flags.action.action_set + assert action_set.multiaction is True + assert action_set.strict is True + assert action_set.demo_mode == "all_blue" + # benchmark's own args are untouched + assert bench.high_level_action_set_args.multiaction is False + assert bench.high_level_action_set_args.demo_mode is None + + def test_set_reproducibility_mode_zeroes_temperature(self): + args = GenericAgentArgs(chat_model_args=FakeModelArgs(temperature=0.7)) + args.set_reproducibility_mode() + assert args.chat_model_args.temperature == 0 + + def test_prepare_and_close_delegate_to_model_args(self): + model_args = FakeModelArgs() + args = GenericAgentArgs(chat_model_args=model_args) + assert args.prepare() == "prepared" + assert args.close() == "closed" + assert model_args.prepared == ["prepare", "close"] + + def test_make_agent(self): + flags = make_flags() + args = GenericAgentArgs(chat_model_args=FakeModelArgs(), flags=flags, max_retry=2) + agent = args.make_agent() + assert isinstance(agent, GenericAgent) + assert agent.flags is flags + assert agent.max_retry == 2 + assert isinstance(agent.chat_llm, FakeChat) + + +# --------------------------------------------------------------------------- GenericAgent + + +class TestGenericAgentInit: + def test_initial_state(self): + agent = make_agent() + assert agent.plan == "No plan yet" + assert agent.plan_step == -1 + assert agent.actions == agent.memories == agent.thoughts == agent.obs_history == [] + assert agent.seed is None + assert "click" in agent.action_set.action_set + + def test_som_without_screenshot_is_disabled(self): + flags = make_flags(obs=dict(use_som=True, use_screenshot=False)) + with pytest.warns(UserWarning, match="use_som=True requires use_screenshot=True"): + make_agent(flags=flags) + assert flags.obs.use_som is False + + def test_screenshot_without_vision_is_disabled(self): + flags = make_flags(obs=dict(use_screenshot=True)) + with pytest.warns(UserWarning, match="does not support vision"): + make_agent(flags=flags, vision_support=False) + assert flags.obs.use_screenshot is False + + def test_screenshot_with_vision_is_kept(self, recwarn): + flags = make_flags(obs=dict(use_screenshot=True, use_som=True)) + make_agent(flags=flags, vision_support=True) + assert flags.obs.use_screenshot is True and flags.obs.use_som is True + assert not [w for w in recwarn if "Disabling" in str(w.message)] + + def test_obs_preprocessor_uses_flags(self, monkeypatch): + seen = {} + + def fake_make(obs_flags): + seen["flags"] = obs_flags + return lambda obs: {**obs, "processed": True} + + monkeypatch.setattr(dp, "make_obs_preprocessor", fake_make) + flags = make_flags() + agent = make_agent(flags=flags) + assert seen["flags"] is flags.obs + assert agent.obs_preprocessor({"a": 1}) == {"a": 1, "processed": True} + + def test_reset_clears_state(self): + agent = make_agent() + agent.get_action(make_obs()) + agent.reset(seed=7) + assert agent.seed == 7 + assert agent.plan == "No plan yet" and agent.plan_step == -1 + assert agent.actions == agent.memories == agent.thoughts == agent.obs_history == [] + + +class TestGetMaxes: + @pytest.mark.parametrize( + "flag_max, total, inp, trunc, expected", + [ + (None, None, None, 20, (None, 20)), + (1000, None, None, 5, (1000, 5)), + (1000, 800, 900, 20, (800, 20)), + (None, 4096, 2048, 0, (2048, 20)), + (None, None, 3000, None, (3000, 20)), + ], + ) + def test_get_maxes(self, flag_max, total, inp, trunc, expected): + flags = make_flags(max_prompt_tokens=flag_max, max_trunc_itr=trunc) + agent = make_agent(flags=flags, max_total_tokens=total, max_input_tokens=inp) + assert agent._get_maxes() == expected + + +class TestGetAction: + def test_happy_path(self): + agent = make_agent() + action, info = agent.get_action(make_obs()) + assert action == "click('12')" + assert info.think == "ok" + assert info.stats["n_retry"] == 0 + assert info.stats["busted_retry"] == 0 + assert info.stats["n_calls"] == 1 + # cost tracker stats are merged in by the decorator + assert {"input_tokens", "output_tokens", "cost"} <= set(info.stats) + assert info.extra_info["chat_model_args"]["model_name"] == "fake/model" + roles = [m["role"] for m in info.chat_messages] + assert roles == ["system", "user", "assistant"] + assert agent.actions == ["click('12')"] + assert agent.thoughts == ["ok"] + assert agent.memories == [None] + assert len(agent.obs_history) == 1 + + def test_prompt_contains_goal_observation_and_history(self): + agent = make_agent( + responses=[ + "firstclick('12')", + "secondclick('13')", + ] + ) + agent.get_action(make_obs(0)) + action, info = agent.get_action(make_obs(1)) + assert action == "click('13')" + user_prompt = str(info.chat_messages[1]) + assert "Click the submit button" in user_prompt + assert "[12] button 'Submit 1'" in user_prompt + assert "# History of interaction with the task" in user_prompt + assert "\nclick('12')\n" in user_prompt + assert "\nfirst\n" in user_prompt + assert info.chat_messages[0]["content"].startswith("You are an agent") + + def test_plan_and_memory_are_tracked(self): + flags = make_flags(use_plan=True, use_memory=True) + agent = make_agent( + flags=flags, + responses=[ + "t1. click1" + "remember 12click('12')" + ], + ) + agent.get_action(make_obs()) + assert agent.plan == "1. click" + assert agent.plan_step == "1" + assert agent.memories == ["remember 12"] + + def test_plan_unchanged_when_not_returned(self): + agent = make_agent() + agent.get_action(make_obs()) + assert agent.plan == "No plan yet" and agent.plan_step == -1 + + def test_retry_after_parse_error(self): + agent = make_agent(responses=["no tags here", "click('12')"]) + action, info = agent.get_action(make_obs()) + assert action == "click('12')" + assert info.stats["n_retry"] == 1 + assert info.stats["busted_retry"] == 0 + assert info.stats["n_calls"] == 2 + roles = [m["role"] for m in info.chat_messages] + assert roles == ["system", "user", "assistant", "user", "assistant"] + assert "Missing the key " in info.chat_messages[3]["content"] + # think falls back to the raw text when its tag is missing + assert info.think == "click('12')" + + def test_busted_retry_returns_none_action(self): + agent = make_agent(responses=["garbage"], max_retry=2) + action, info = agent.get_action(make_obs()) + assert action is None + assert info.stats["busted_retry"] == 1 + assert info.stats["n_retry"] == 3 + assert info.stats["n_calls"] == 2 + assert agent.actions == [None] + assert agent.thoughts == [None] + + def test_fit_tokens_called_with_maxes_and_system_prompt(self, monkeypatch): + real_fit = dp.fit_tokens + spy = mock.Mock(side_effect=real_fit) + monkeypatch.setattr(dp, "fit_tokens", spy) + flags = make_flags(max_prompt_tokens=None, max_trunc_itr=7) + monkeypatch.setattr(dp, "count_tokens", lambda text, model=None: 1) + agent = make_agent(flags=flags, max_input_tokens=5000) + agent.get_action(make_obs()) + kwargs = spy.call_args.kwargs + assert kwargs["max_prompt_tokens"] == 5000 + assert kwargs["max_iterations"] == 7 + assert kwargs["model_name"] == "fake/model" + assert kwargs["additional_prompts"]["role"] == "system" + assert isinstance(kwargs["shrinkable"], MainPrompt) + + def test_prompt_is_shrunk_to_fit(self, monkeypatch): + monkeypatch.setattr(dp, "count_tokens", lambda text, model=None: len(text.split())) + big_html = "\n".join(f"

line {i}

" for i in range(400)) + flags = make_flags( + obs=dict(use_ax_tree=False), max_prompt_tokens=700, use_concrete_example=False + ) + agent = make_agent(flags=flags) + _, info = agent.get_action(make_obs(pruned_html=big_html)) + prompt = str(info.chat_messages[1]) + assert "lines to reduce prompt size" in prompt + assert len(prompt.split()) <= 700 + + +# --------------------------------------------------------------------------- generic_agent_prompt + + +def _main_prompt(flags, obs=None, **kwargs): + action_set = flags.action.action_set.make_action_set() + params = dict( + action_set=action_set, + obs_history=[obs or make_obs()], + actions=[], + memories=[], + thoughts=[], + previous_plan="No plan yet", + step=-1, + flags=flags, + ) + params.update(kwargs) + return MainPrompt(**params) + + +class TestMainPrompt: + def test_goal_mode_warns_on_multiple_user_messages(self, caplog): + msgs = [ + {"role": "user", "message": "a", "timestamp": 0}, + {"role": "user", "message": "b", "timestamp": 1}, + ] + with caplog.at_level(logging.WARNING): + mp = _main_prompt(make_flags(), obs=make_obs(chat_messages=msgs)) + assert "Consider switching to `enable_chat=True`" in caplog.text + assert isinstance(mp.instructions, dp.GoalInstructions) + + def test_goal_mode_single_message_no_warning(self, caplog): + with caplog.at_level(logging.WARNING): + _main_prompt(make_flags()) + assert "enable_chat" not in caplog.text + + def test_chat_mode_uses_chat_instructions(self): + flags = make_flags(enable_chat=True, extra_instructions="be brief") + mp = _main_prompt(flags) + assert isinstance(mp.instructions, dp.ChatInstructions) + text = str(mp.prompt) + assert "## Chat messages:" in text and "Click submit" in text + assert "be brief" in text + + def test_abstract_and_concrete_examples(self): + flags = make_flags( + use_abstract_example=True, + use_concrete_example=True, + use_plan=True, + use_memory=True, + use_criticise=True, + ) + text = str(_main_prompt(flags).prompt) + assert "# Abstract Example" in text + assert "# Concrete Example" in text + for tag in ("", "", "", "", "", ""): + assert text.count(tag) >= 2, tag + assert "You just executed step -1 of the previously proposed plan" in text + + def test_examples_disabled(self): + flags = make_flags(use_abstract_example=False, use_concrete_example=False) + text = str(_main_prompt(flags).prompt) + assert "Abstract Example" not in text and "Concrete Example" not in text + + @pytest.mark.parametrize( + "be_cautious, multiaction, visible", + [(True, True, True), (True, False, False), (False, True, False)], + ) + def test_be_cautious_only_in_multiaction(self, be_cautious, multiaction, visible): + flags = make_flags( + be_cautious=be_cautious, + action=dict( + action_set=HighLevelActionSetArgs(subsets=["bid"], multiaction=multiaction) + ), + ) + assert ("Be very cautious" in str(_main_prompt(flags).prompt)) is visible + + def test_hints_flag(self): + assert "Note:\n* Some tasks" in str(_main_prompt(make_flags(use_hints=True)).prompt) + assert "Some tasks may be game" not in str(_main_prompt(make_flags()).prompt) + + def test_screenshot_is_appended_when_enabled(self, monkeypatch): + monkeypatch.setattr(dp, "image_to_jpg_base64_url", lambda img: "data:image/jpeg;base64,Q") + flags = make_flags(obs=dict(use_screenshot=True)) + prompt = _main_prompt(flags, obs=make_obs(screenshot="IMG")).prompt + assert isinstance(prompt, HumanMessage) + assert prompt["content"][-1]["image_url"]["url"] == "data:image/jpeg;base64,Q" + + def test_shrink_delegates_to_history_and_obs(self): + mp = _main_prompt(make_flags()) + with mock.patch.object(mp.history, "shrink") as h, mock.patch.object(mp.obs, "shrink") as o: + mp.shrink() + h.assert_called_once_with() + o.assert_called_once_with() + + def test_parse_answer_combines_all_elements(self): + flags = make_flags(use_plan=True, use_memory=True, use_criticise=True) + mp = _main_prompt(flags) + ans = mp._parse_answer( + "TP3M" + "DCclick('12')" + ) + assert ans == { + "think": "T", + "plan": "P", + "step": "3", + "memory": "M", + "action_draft": "D", + "criticise": "C", + "action": "click('12')", + } + + def test_parse_answer_hidden_elements_ignored(self): + flags = make_flags(use_thinking=False) + ans = _main_prompt(flags)._parse_answer( + "TPclick('12')" + ) + assert ans == {"action": "click('12')"} + + def test_parse_answer_raises_on_invalid_action(self): + with pytest.raises(ParseError): + _main_prompt(make_flags())._parse_answer("not_an_action") + + +class TestAnswerElements: + def test_memory_optional_and_merged(self): + assert Memory().parse_answer("nothing") == {} + assert Memory().parse_answer("ab") == {"memory": "a\nb"} + + def test_plan_parse_and_prompt(self): + plan = Plan("1. do", 2) + assert "You just executed step 2" in plan.prompt + assert "1. do" in plan.prompt + assert plan.parse_answer("x4") == {"plan": "x", "step": "4"} + assert plan.parse_answer("no plan") == {} + assert Plan("p", 0, visible=False).parse_answer("x") == {} + + def test_plan_rejects_duplicate_tags(self): + with pytest.raises(ParseError, match="multiple instances of the key plan"): + Plan("p", 0).parse_answer("ab") + + def test_criticise_parse(self): + crit = Criticise() + assert crit.prompt == "" + assert crit.parse_answer("dc") == { + "action_draft": "d", + "criticise": "c", + } + assert crit.parse_answer("nothing") == {} + + def test_flags_asdict_roundtrip(self): + flags = make_flags(use_plan=True, extra_instructions="x") + d = flags.asdict() + assert d["use_plan"] is True + assert d["obs"]["use_html"] is True + assert d["extra_instructions"] == "x" + clone = flags.copy() + clone.obs.use_html = False + assert flags.obs.use_html is True diff --git a/tests/agents/test_reproducibility_agent_unit.py b/tests/agents/test_reproducibility_agent_unit.py new file mode 100644 index 000000000..f5328a655 --- /dev/null +++ b/tests/agents/test_reproducibility_agent_unit.py @@ -0,0 +1,357 @@ +"""Unit tests for agentlab.agents.generic_agent.reproducibility_agent (hermetic, no LLM).""" + +import json +import shutil +from dataclasses import dataclass +from pathlib import Path +from unittest import mock + +import pytest +from bgym import HighLevelActionSetArgs +from browsergym.experiments.agent import AgentInfo + +from agentlab.agents import dynamic_prompting as dp +from agentlab.agents.generic_agent import reproducibility_agent as ra +from agentlab.agents.generic_agent.generic_agent import GenericAgentArgs +from agentlab.agents.generic_agent.generic_agent_prompt import GenericPromptFlags +from agentlab.experiments.loop import ExpResult, StepInfo +from agentlab.llm.base_api import BaseModelArgs +from agentlab.llm.llm_utils import Discussion, HumanMessage, SystemMessage + +TEST_STUDY = Path(__file__).parent.parent / "data" / "test_study" +EXP_WITH_SUMMARY = "2024-08-01_10-20-52_GenericAgent_on_miniwob.ascending-numbers_64_e6d2d5" + + +@pytest.fixture(autouse=True) +def no_sleep(monkeypatch): + sleep = mock.Mock() + monkeypatch.setattr(ra.time, "sleep", sleep) + return sleep + + +@dataclass +class NoLLMArgs(BaseModelArgs): + model_name: str = "fake/repro" + + def make_model(self): + def _fail(messages): + raise AssertionError("ReproAgent must not call the original model") + + return _fail + + +def make_flags(): + return GenericPromptFlags( + obs=dp.ObsFlags(use_html=True, use_ax_tree=True, use_screenshot=False), + action=dp.ActionFlags(action_set=HighLevelActionSetArgs(subsets=["bid"])), + use_thinking=True, + use_concrete_example=False, + ) + + +def make_obs(): + return { + "goal_object": [{"type": "text", "text": "Click the submit button"}], + "chat_messages": [], + "pruned_html": "", + "axtree_txt": "[12] button 'Submit'", + "last_action_error": "", + "focused_element_bid": "", + "open_pages_urls": ["https://example.com"], + "open_pages_titles": ["Example"], + "active_page_index": 0, + } + + +def write_step(exp_dir, step, chat_messages=None, action=None, with_chat=True): + agent_info = {"chat_messages": chat_messages} if with_chat else {} + StepInfo(step=step, obs=None, action=action, agent_info=agent_info).save_step_info( + exp_dir, save_screenshot=False + ) + + +def old_messages(answer="oldclick('12')"): + return [ + {"role": "system", "content": "old system prompt"}, + {"role": "user", "content": "old user prompt"}, + {"role": "assistant", "content": answer}, + ] + + +def make_repro_agent(exp_dir): + args = ra.ReproAgentArgs( + chat_model_args=NoLLMArgs(), flags=make_flags(), max_retry=2, _repro_dir=exp_dir + ) + return args.make_agent() + + +# --------------------------------------------------------------------------- ReproChatModel + + +class TestReproChatModel: + def test_replays_next_message_and_sleeps(self, no_sleep): + msgs = old_messages() + model = ra.ReproChatModel(msgs, delay=3) + out = model([msgs[0], msgs[1]]) + assert out is msgs[2] + no_sleep.assert_called_once_with(3) + assert model.new_messages == [msgs[0], msgs[1], msgs[2]] + assert model.get_stats() == {} + + def test_input_list_is_not_mutated(self): + msgs = old_messages() + sent = [msgs[0], msgs[1]] + ra.ReproChatModel(msgs, delay=0)(sent) + assert len(sent) == 2 + + def test_missing_response_returns_none_action_without_sleeping(self, no_sleep): + msgs = old_messages()[:2] + model = ra.ReproChatModel(msgs) + out = model(list(msgs)) + assert out == {"role": "assistant", "content": "None"} + no_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- ReproAgentArgs + + +class TestReproAgentArgs: + def test_name_prefixed(self): + args = ra.ReproAgentArgs(chat_model_args=NoLLMArgs(model_name="a/b")) + assert args.agent_name == "Repro_GenericAgent-a_b" + + def test_missing_model_args_does_not_raise(self): + args = ra.ReproAgentArgs() + assert args.agent_name == "Repro_None" + + def test_make_agent_binds_exp_dir(self, tmp_path): + agent = make_repro_agent(tmp_path) + assert isinstance(agent, ra.ReproAgent) + assert isinstance(agent.exp_result, ExpResult) + assert agent.exp_result.exp_dir == tmp_path + assert agent.max_retry == 2 + + +# --------------------------------------------------------------------------- ReproAgent.get_action + + +class TestReproAgentGetAction: + def test_replays_recorded_answer_and_diffs(self, tmp_path): + write_step(tmp_path, 0, chat_messages=old_messages()) + agent = make_repro_agent(tmp_path) + action, info = agent.get_action(make_obs()) + assert action == "click('12')" + assert info.think == "old" + assert isinstance(agent.chat_llm, ra.ReproChatModel) + assert "Old Version" in info.html_page and "New Version" in info.html_page + assert "table.diff" in info.html_page + # the prompts differ (old one was a placeholder), so some lines changed + assert info.stats["lines_added"] > 0 + assert info.stats["lines_removed"] > 0 + assert 0 < info.stats["difference_ratio"] <= 1 + assert agent.actions == ["click('12')"] + + def test_uses_step_index_from_action_count(self, tmp_path): + write_step(tmp_path, 0, chat_messages=old_messages()) + write_step( + tmp_path, 1, chat_messages=old_messages("2ndclick('7')") + ) + agent = make_repro_agent(tmp_path) + agent.get_action(make_obs()) + action, info = agent.get_action(make_obs()) + assert action == "click('7')" + assert info.think == "2nd" + + @pytest.mark.xfail( + strict=True, + reason="BUG: ReproChatModel shallow-copies the Discussion, so the replayed answer is " + "appended twice (n_retry=0.5 and a duplicated assistant message)", + ) + def test_replay_does_not_duplicate_answer(self, tmp_path): + write_step(tmp_path, 0, chat_messages=old_messages()) + agent = make_repro_agent(tmp_path) + _, info = agent.get_action(make_obs()) + assert info.stats["n_retry"] == 0 + assert [m["role"] for m in info.chat_messages] == ["system", "user", "assistant"] + + def test_no_chat_messages_returns_error_page(self, tmp_path): + write_step(tmp_path, 0, with_chat=False) + (tmp_path / "summary_info.json").write_text(json.dumps({"err_msg": "env crashed"})) + agent = make_repro_agent(tmp_path) + action, info = agent.get_action(make_obs()) + assert action is None + assert isinstance(info, AgentInfo) + assert "Agent had no chat messages" in info.markdown_page + assert "env crashed" in info.markdown_page + assert agent.actions == [] + + def test_missing_response_rebuilt_from_recorded_action(self, tmp_path): + write_step(tmp_path, 0, chat_messages=old_messages()[:2], action="click('99')") + agent = make_repro_agent(tmp_path) + action, _ = agent.get_action(make_obs()) + assert action == "click('99')" + assert agent.chat_llm.old_messages[2] == { + "role": "assistant", + "content": "click('99')", + } + + def test_missing_response_without_recorded_action_yields_none(self, tmp_path): + write_step(tmp_path, 0, chat_messages=old_messages()[:2], action=None) + agent = make_repro_agent(tmp_path) + action, info = agent.get_action(make_obs()) + assert action is None + assert info.stats["busted_retry"] == 0 + + +# --------------------------------------------------------------------------- helpers + + +class TestMakeAgentStats: + def test_converts_dict_agent_info_and_adds_diff(self): + old = [{"role": "user", "content": "a\nb"}] + new = [{"role": "user", "content": "a\nc"}] + action, info = ra._make_agent_stats("noop()", {"stats": {"k": 1}}, None, old, new) + assert action == "noop()" + assert isinstance(info, AgentInfo) + assert info.stats["k"] == 1 + assert info.stats["lines_added"] == 1 and info.stats["lines_removed"] == 1 + assert " message:\nhi\n" + + def test_discussion_uses_to_string(self): + disc = Discussion([SystemMessage("S"), HumanMessage("U")]) + assert ra._format_messages(disc) == "S\nU" + + def test_unknown_type_raises(self): + with pytest.raises(ValueError, match="Unknown message type"): + ra._format_messages([42]) + + +class TestMakeBackwardCompatible: + def _args(self, action_set, multi_actions=None): + flags = make_flags() + flags.action.action_set = action_set + flags.action.multi_actions = multi_actions + return GenericAgentArgs(chat_model_args=NoLLMArgs(), flags=flags) + + @pytest.mark.parametrize( + "legacy, subsets", + [("bid", ["bid"]), ("bid+nav", ["bid", "nav"]), (["bid", "tab"], ["bid", "tab"])], + ) + def test_legacy_action_set_converted(self, legacy, subsets): + args = ra._make_backward_compatible(self._args(legacy, multi_actions=True)) + action_set = args.flags.action.action_set + assert isinstance(action_set, HighLevelActionSetArgs) + assert list(action_set.subsets) == subsets + assert action_set.multiaction is True + + def test_modern_action_set_untouched(self): + modern = HighLevelActionSetArgs(subsets=["bid"]) + args = ra._make_backward_compatible(self._args(modern)) + assert args.flags.action.action_set is modern + + +class TestDiffHelpers: + def test_diff_stats_identical(self): + assert ra._diff_stats("a\nb", "a\nb") == dict( + lines_added=0, lines_removed=0, difference_ratio=0.0 + ) + + def test_diff_stats_counts(self): + stats = ra._diff_stats("a\nb\nc", "a\nx\nc\nd") + assert stats == dict(lines_added=2, lines_removed=1, difference_ratio=3 / 8) + + @pytest.mark.xfail( + strict=True, reason="BUG: _diff_stats divides by zero when both strings are empty" + ) + def test_diff_stats_empty_strings(self): + assert ra._diff_stats("", "")["difference_ratio"] == 0 + + def test_make_diff_strips_nowrap_and_injects_style(self): + page = ra._make_diff("same\noldline", "same\nnewline") + assert 'nowrap="nowrap"' not in page + assert "Old Version" in page and "New Version" in page + assert ".diff_add {" in page + assert "oldline" in page and "newline" in page + + def test_set_style_appends_by_default(self): + html = "" + out = ra._set_style(html, "p{}") + assert "" in out + + def test_set_style_prepend(self): + html = "" + out = ra._set_style(html, "p{}", prepend_previous_style=True) + assert "" in out + + def test_set_style_creates_tag_when_missing(self): + out = ra._set_style("", "p{}") + assert "" in out + + +# --------------------------------------------------------------------------- study-level + + +class TestMakeReproAgent: + def test_from_fixture_study(self): + exp_dir = TEST_STUDY / EXP_WITH_SUMMARY + agent_args = ExpResult(exp_dir).exp_args.agent_args + repro = ra.make_repro_agent(agent_args, exp_dir=str(exp_dir)) + assert isinstance(repro, ra.ReproAgentArgs) + assert repro._repro_dir == exp_dir + assert repro.chat_model_args is agent_args.chat_model_args + assert repro.flags is agent_args.flags + assert repro.max_retry == agent_args.max_retry + assert repro.agent_name == f"Repro_GenericAgent-{agent_args.chat_model_args.model_name}" + + def test_rejects_non_generic_agent_args(self, tmp_path): + with pytest.raises(AssertionError): + ra.make_repro_agent(object(), exp_dir=tmp_path) + + def test_rejects_missing_dir(self, tmp_path): + args = GenericAgentArgs(chat_model_args=NoLLMArgs(), flags=make_flags()) + with pytest.raises(AssertionError): + ra.make_repro_agent(args, exp_dir=tmp_path / "missing") + + +class TestReproduceStudy: + def test_builds_one_repro_exp_per_original_exp(self, tmp_path, monkeypatch): + study_dir = tmp_path / "study" + shutil.copytree(TEST_STUDY, study_dir) + + created = {} + + class FakeStudy: + def __init__(self, benchmark, agent_args): + created["benchmark"] = benchmark + created["agent_args"] = agent_args + self.exp_args_list = None + + monkeypatch.setattr(ra, "Study", FakeStudy) + study = ra.reproduce_study(study_dir, log_level=10) + + assert isinstance(study, FakeStudy) + assert created["benchmark"] == "miniwob" + assert len(study.exp_args_list) == 2 + exp_dirs = set() + for exp_args in study.exp_args_list: + agent_args = exp_args.agent_args + assert isinstance(agent_args, ra.ReproAgentArgs) + assert agent_args.agent_name.startswith("Repro_GenericAgent-") + # legacy "bid" string action set was upgraded + assert isinstance(agent_args.flags.action.action_set, HighLevelActionSetArgs) + assert exp_args.logging_level == 10 + assert exp_args.env_args.task_name == "miniwob.ascending-numbers" + exp_dirs.add(Path(agent_args._repro_dir).name) + assert exp_dirs == {p.name for p in TEST_STUDY.iterdir() if p.is_dir()} + assert created["agent_args"][0] is study.exp_args_list[-1].agent_args