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