Spaces:
Running
Running
| import os | |
| import unittest | |
| from pathlib import Path | |
| from tempfile import TemporaryDirectory | |
| from unittest.mock import patch | |
| import demo | |
| import action_mapping | |
| import curriculum | |
| import agents | |
| import observation | |
| import slm_policy | |
| import trainer | |
| from reward import RewardEngine | |
| from server.browser_env_environment import BrowserEnvironment | |
| from models import BrowserAction, BrowserElement, BrowserObservation, ConstraintState, RewardBreakdown | |
| class PhaseFourAndFiveTests(unittest.TestCase): | |
| def test_compact_elements_with_stats_prioritizes_instruction_match(self): | |
| raw_obs = { | |
| "goal": 'Click "Submit".', | |
| "elements": [ | |
| {"id": "1", "role": "button", "tag": "button", "text": "Cancel", "visible": True}, | |
| {"id": "2", "role": "button", "tag": "button", "text": "Submit", "visible": True}, | |
| {"id": "3", "role": "textbox", "tag": "input", "text": "", "visible": True}, | |
| ], | |
| } | |
| elements, stats = observation.compact_elements_with_stats(raw_obs, {"instruction": 'Click "Submit".', "max_elements": 2}) | |
| self.assertEqual(elements[0].id, "2") | |
| self.assertIn("submit_like", elements[0].attributes["semantic_hints"]) | |
| self.assertEqual(stats["elements_before"], 3) | |
| self.assertEqual(stats["elements_after"], 2) | |
| def test_compact_elements_respects_hidden_attrs_in_text_fallback(self): | |
| raw_obs = { | |
| "goal": "Fill the form.", | |
| "elements": [ | |
| { | |
| "id": "email", | |
| "role": "textbox", | |
| "tag": "input", | |
| "text": "", | |
| "value": "secret@example.com", | |
| "placeholder": "Email", | |
| "title": "Work email", | |
| "visible": True, | |
| } | |
| ], | |
| } | |
| elements, _ = observation.compact_elements_with_stats( | |
| raw_obs, | |
| { | |
| "instruction": "Fill the form.", | |
| "max_elements": 5, | |
| "hidden_attributes": ["value", "placeholder", "title"], | |
| }, | |
| ) | |
| self.assertEqual(len(elements), 1) | |
| self.assertEqual(elements[0].text, "") | |
| def test_translate_browser_action_uses_semantic_target_fallback(self): | |
| elements = [ | |
| BrowserElement( | |
| id="cancel-btn", | |
| role="button", | |
| tag="button", | |
| text="Cancel", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True, "rank_score": 0.8}, | |
| ), | |
| BrowserElement( | |
| id="submit-btn", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True, "rank_score": 4.2, "semantic_hints": ["clickable", "submit_like"]}, | |
| ), | |
| ] | |
| result = action_mapping.translate_browser_action( | |
| BrowserAction(action_type="click", target_id="999", reasoning="Click the Submit button"), | |
| elements, | |
| instruction='Click "Submit".', | |
| ) | |
| self.assertEqual(result.resolved_action.target_id, "submit-btn") | |
| self.assertEqual(result.metadata["mode"], "semantic") | |
| self.assertEqual(result.browsergym_action, "click('submit-btn')") | |
| def test_find_submit_target_ignores_hidden_or_disabled_controls(self): | |
| elements = [ | |
| BrowserElement( | |
| id="hidden-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=False, | |
| attributes={"clickable": True}, | |
| ), | |
| BrowserElement( | |
| id="disabled-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=False, | |
| visible=True, | |
| attributes={"clickable": True}, | |
| ), | |
| BrowserElement( | |
| id="active-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True}, | |
| ), | |
| ] | |
| target = action_mapping.find_submit_target(elements) | |
| self.assertEqual(target, "active-submit") | |
| def test_rank_elements_preserves_instruction_anchor_under_topk(self): | |
| elements = [ | |
| BrowserElement( | |
| id="e1", | |
| role="button", | |
| tag="button", | |
| text="Continue to next screen now", | |
| visible=True, | |
| enabled=True, | |
| attributes={"clickable": True}, | |
| ), | |
| BrowserElement( | |
| id="e2", | |
| role="button", | |
| tag="button", | |
| text="Green", | |
| visible=True, | |
| enabled=True, | |
| attributes={"clickable": True}, | |
| ), | |
| ] | |
| ranked, stats = observation._rank_elements( # type: ignore[attr-defined] | |
| elements, | |
| 'Click "Green" to continue.', | |
| history_texts=[], | |
| max_elements=1, | |
| ) | |
| self.assertEqual(len(ranked), 1) | |
| self.assertEqual(ranked[0].id, "e2") | |
| self.assertGreaterEqual(int(stats.get("anchor_recall_boost", 0)), 0) | |
| def test_collect_environment_status_flags_missing_miniwob(self): | |
| with patch.dict(os.environ, {}, clear=True): | |
| status = trainer.collect_environment_status() | |
| self.assertFalse(status["miniwob_ready"]) | |
| self.assertTrue(status["llm_services_enabled"]) | |
| self.assertFalse(status["openai_api_key_present"]) | |
| def test_collect_environment_status_reports_openai_model_for_openai_provider(self): | |
| with patch.dict( | |
| os.environ, | |
| { | |
| "BROWSER_ENV_LLM_PROVIDER": "openai_compatible", | |
| "OPENAI_MODEL": "openai/gpt-oss-20b", | |
| }, | |
| clear=True, | |
| ): | |
| status = trainer.collect_environment_status() | |
| self.assertEqual(status["llm_model_id"], "openai/gpt-oss-20b") | |
| def test_validate_runtime_raises_clear_message(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=True): | |
| with self.assertRaises(RuntimeError) as exc: | |
| trainer.validate_runtime() | |
| self.assertIn("MINIWOB_URL", str(exc.exception)) | |
| def test_oracle_json_fallback_recovers_truncated_action(self): | |
| text = ( | |
| '{\n' | |
| ' "action_type": "click",\n' | |
| ' "target_id": "14",\n' | |
| ' "text": null,\n' | |
| " \"browsergym_action\": \"click('14')\",\n" | |
| ' "confidence": 0.9,\n' | |
| ' "reasoning": "truncated' | |
| ) | |
| action = agents._extract_json(text, action_fallback=True) | |
| self.assertEqual(action["action_type"], "click") | |
| self.assertEqual(action["target_id"], "14") | |
| self.assertAlmostEqual(action["confidence"], 0.9) | |
| self.assertNotIn("browsergym_action", action) | |
| def test_sanitize_oracle_action_keeps_select_target_without_clickable_flag(self): | |
| elements = [ | |
| { | |
| "id": "country-select", | |
| "role": "combobox", | |
| "tag": "select", | |
| "type": "select-one", | |
| "text": "Country", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": False, | |
| "options_preview": "India | USA", | |
| }, | |
| { | |
| "id": "submit-btn", | |
| "role": "button", | |
| "tag": "button", | |
| "type": "", | |
| "text": "Submit", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": True, | |
| "options_preview": "", | |
| }, | |
| ] | |
| sanitized = agents._sanitize_oracle_action( | |
| { | |
| "action_type": "select", | |
| "target_id": "country-select", | |
| "text": "India", | |
| }, | |
| elements, | |
| ) | |
| self.assertEqual(sanitized["action_type"], "select") | |
| self.assertEqual(sanitized["target_id"], "country-select") | |
| def test_sanitize_oracle_action_retargets_hidden_click_target(self): | |
| elements = [ | |
| { | |
| "id": "hidden-submit", | |
| "role": "button", | |
| "tag": "button", | |
| "type": "", | |
| "text": "Submit", | |
| "visible": False, | |
| "enabled": True, | |
| "clickable": True, | |
| "options_preview": "", | |
| }, | |
| { | |
| "id": "visible-submit", | |
| "role": "button", | |
| "tag": "button", | |
| "type": "", | |
| "text": "Submit", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": True, | |
| "options_preview": "", | |
| }, | |
| ] | |
| sanitized = agents._sanitize_oracle_action( | |
| { | |
| "action_type": "click", | |
| "target_id": "hidden-submit", | |
| }, | |
| elements, | |
| ) | |
| self.assertEqual(sanitized["action_type"], "click") | |
| self.assertEqual(sanitized["target_id"], "visible-submit") | |
| def test_sanitize_oracle_action_accepts_semantic_click_target_without_clickable_attr(self): | |
| elements = [ | |
| { | |
| "id": "semantic-button", | |
| "role": "button", | |
| "tag": "button", | |
| "type": "", | |
| "text": "Continue", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": False, | |
| "options_preview": "", | |
| } | |
| ] | |
| sanitized = agents._sanitize_oracle_action( | |
| { | |
| "action_type": "click", | |
| "target_id": "semantic-button", | |
| }, | |
| elements, | |
| ) | |
| self.assertEqual(sanitized["action_type"], "click") | |
| self.assertEqual(sanitized["target_id"], "semantic-button") | |
| def test_sanitize_oracle_action_fallback_uses_semantic_click_target(self): | |
| elements = [ | |
| { | |
| "id": "semantic-link", | |
| "role": "link", | |
| "tag": "a", | |
| "type": "", | |
| "text": "Next", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": False, | |
| "options_preview": "", | |
| }, | |
| { | |
| "id": "non-interactive", | |
| "role": "generic", | |
| "tag": "div", | |
| "type": "", | |
| "text": "Container", | |
| "visible": True, | |
| "enabled": True, | |
| "clickable": False, | |
| "options_preview": "", | |
| }, | |
| ] | |
| sanitized = agents._sanitize_oracle_action( | |
| { | |
| "action_type": "click", | |
| "target_id": "missing-target", | |
| }, | |
| elements, | |
| ) | |
| self.assertEqual(sanitized["action_type"], "click") | |
| self.assertEqual(sanitized["target_id"], "semantic-link") | |
| def test_translate_browser_action_ignores_hidden_click_candidate_when_target_missing(self): | |
| elements = [ | |
| BrowserElement( | |
| id="hidden-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit now", | |
| enabled=True, | |
| visible=False, | |
| attributes={"clickable": True, "rank_score": 8.0}, | |
| ), | |
| BrowserElement( | |
| id="visible-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True, "rank_score": 4.0}, | |
| ), | |
| ] | |
| result = action_mapping.translate_browser_action( | |
| BrowserAction(action_type="click", target_id="unknown", reasoning="click submit"), | |
| elements, | |
| instruction='Click "Submit".', | |
| ) | |
| self.assertEqual(result.resolved_action.target_id, "visible-submit") | |
| def test_full_curriculum_falls_back_to_curated_when_registry_is_unavailable(self): | |
| with TemporaryDirectory() as tmpdir: | |
| root = Path(tmpdir) | |
| for name in ("click-test", "enter-text", "book-flight", "index"): | |
| (root / f"{name}.html").write_text("", encoding="utf-8") | |
| with patch.dict( | |
| os.environ, | |
| { | |
| "MINIWOB_URL": f"file://{root}/", | |
| "BROWSER_ENV_CURRICULUM_MODE": "miniwob_full", | |
| "BROWSER_ENV_CURRICULUM_MAX_TASKS": "0", | |
| }, | |
| clear=True, | |
| ): | |
| pool = curriculum.CurriculumPool() | |
| task_ids = {variant.task_id for variant in pool.variants} | |
| self.assertGreaterEqual(len(pool.variants), 3) | |
| self.assertIn("browsergym/miniwob.click-test", task_ids) | |
| self.assertIn("browsergym/miniwob.enter-text", task_ids) | |
| self.assertIn("browsergym/miniwob.book-flight", task_ids) | |
| self.assertIn("browsergym/miniwob.choose-list", task_ids) | |
| def test_full_curriculum_does_not_use_html_fallback_without_opt_in(self): | |
| with TemporaryDirectory() as tmpdir: | |
| root = Path(tmpdir) | |
| local_only = ("zz-local-only-task", "zz-second-local-only-task") | |
| for name in local_only: | |
| (root / f"{name}.html").write_text("", encoding="utf-8") | |
| with patch.dict( | |
| os.environ, | |
| { | |
| "MINIWOB_URL": f"file://{root}/", | |
| "BROWSER_ENV_CURRICULUM_MODE": "miniwob_full", | |
| "BROWSER_ENV_CURRICULUM_MAX_TASKS": "0", | |
| }, | |
| clear=True, | |
| ): | |
| task_ids = curriculum._discover_miniwob_task_ids() | |
| self.assertNotIn("browsergym/miniwob.zz-local-only-task", task_ids) | |
| self.assertNotIn("browsergym/miniwob.zz-second-local-only-task", task_ids) | |
| def test_curriculum_can_reset_to_explicit_preset_variant(self): | |
| pool = curriculum.CurriculumPool() | |
| selected = pool.variant_for_reset(variant_id="medium-select") | |
| self.assertEqual(selected.variant_id, "medium-select") | |
| self.assertEqual(selected.task_id, "browsergym/miniwob.choose-list") | |
| self.assertEqual(selected.seed, 31) | |
| def test_curriculum_ranks_dom_tasks_before_visual_pointer_tasks(self): | |
| variants = [ | |
| curriculum.TaskVariant("browsergym/miniwob.bisect-angle", "visual_pointer", "hard", "hard-bisect-angle"), | |
| curriculum.TaskVariant("browsergym/miniwob.enter-text", "form", "easy", "easy-enter-text"), | |
| curriculum.TaskVariant("browsergym/miniwob.click-button", "click", "easy", "easy-click-button"), | |
| curriculum.TaskVariant("browsergym/miniwob.choose-list", "select", "medium", "medium-choose-list"), | |
| curriculum.TaskVariant("browsergym/miniwob.circle-center", "visual_pointer", "hard", "hard-circle-center"), | |
| ] | |
| ordered = [variant.task_id for variant in sorted(variants, key=curriculum._variant_sort_key)] | |
| self.assertEqual(ordered[0], "browsergym/miniwob.click-button") | |
| self.assertEqual(ordered[1], "browsergym/miniwob.enter-text") | |
| self.assertEqual(ordered[2], "browsergym/miniwob.choose-list") | |
| self.assertTrue(ordered[-1].endswith("circle-center")) | |
| def test_compute_deltas_returns_metric_differences(self): | |
| before = { | |
| "success_rate": 0.25, | |
| "avg_reward": -1.0, | |
| "avg_steps": 8.0, | |
| "avg_oracle_calls": 1.0, | |
| "avg_invalid_actions": 2.0, | |
| } | |
| after = { | |
| "success_rate": 0.75, | |
| "avg_reward": 4.5, | |
| "avg_steps": 5.0, | |
| "avg_oracle_calls": 0.5, | |
| "avg_invalid_actions": 0.0, | |
| } | |
| deltas = trainer.compute_deltas(before, after) | |
| self.assertAlmostEqual(deltas["success_rate"], 0.5) | |
| self.assertAlmostEqual(deltas["avg_reward"], 5.5) | |
| self.assertAlmostEqual(deltas["avg_steps"], -3.0) | |
| self.assertAlmostEqual(deltas["avg_oracle_calls"], -0.5) | |
| self.assertAlmostEqual(deltas["avg_invalid_actions"], -2.0) | |
| def test_reward_engine_adds_progress_credit(self): | |
| breakdown = RewardEngine().compute( | |
| browsergym_reward=0.0, | |
| success=False, | |
| step_delta=1, | |
| progress_delta=1, | |
| oracle_delta=0, | |
| mistake_delta=0, | |
| repetition_delta=0, | |
| ) | |
| self.assertAlmostEqual(breakdown.step_penalty, -0.1) | |
| self.assertAlmostEqual(breakdown.progress_reward, 0.1) | |
| self.assertAlmostEqual(breakdown.efficiency, 0.0) | |
| self.assertAlmostEqual(breakdown.total, 0.0) | |
| def test_reward_engine_populates_rubric_components(self): | |
| breakdown = RewardEngine().compute( | |
| browsergym_reward=1.5, | |
| success=True, | |
| step_delta=2, | |
| progress_delta=1, | |
| oracle_delta=1, | |
| mistake_delta=1, | |
| repetition_delta=1, | |
| judge_quality_reward=0.75, | |
| delayed_penalty=-0.25, | |
| ) | |
| self.assertAlmostEqual(breakdown.task_completion, 11.5) | |
| self.assertAlmostEqual(breakdown.action_validity, -0.5) | |
| self.assertAlmostEqual(breakdown.efficiency, -0.35) | |
| self.assertAlmostEqual(breakdown.non_repetition, -0.2) | |
| self.assertAlmostEqual(breakdown.help_independence, -1.0) | |
| self.assertAlmostEqual(breakdown.trajectory_quality, 0.75) | |
| self.assertAlmostEqual( | |
| breakdown.total, | |
| breakdown.task_completion | |
| + breakdown.action_validity | |
| + breakdown.efficiency | |
| + breakdown.non_repetition | |
| + breakdown.help_independence | |
| + breakdown.trajectory_quality, | |
| ) | |
| def test_adaptive_step_budget_generator_only_mutates_on_triggered_failures(self): | |
| generator = agents.AdaptiveStepBudgetGenerator() | |
| base = { | |
| "task_id": "browsergym/miniwob.click-button-sequence", | |
| "task_family": "click", | |
| "difficulty": "easy", | |
| "variant_id": "easy-click-button-sequence", | |
| "seed": 22, | |
| "max_steps": 12, | |
| } | |
| unchanged = generator.mutate_task(base, {"reason": "submission_failed", "count": 3}) | |
| mutated = generator.mutate_task(base, {"reason": "max_steps_exceeded", "count": 3}) | |
| self.assertEqual(unchanged["max_steps"], 12) | |
| self.assertEqual(mutated["max_steps"], 14) | |
| self.assertEqual(mutated["base_max_steps"], 12) | |
| self.assertEqual(mutated["step_budget_cap"], 20) | |
| self.assertEqual(mutated["step_budget_adaptation_count"], 1) | |
| self.assertEqual(mutated["mutation_kind"], "adaptive_step_budget") | |
| self.assertIn("steps-14", mutated["variant_id"]) | |
| def test_build_agent_bundle_keeps_step_generator_when_llm_disabled(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| bundle = agents.build_agent_bundle() | |
| mutated = bundle.mutate_task( | |
| { | |
| "task_id": "browsergym/miniwob.click-checkboxes", | |
| "task_family": "click", | |
| "difficulty": "medium", | |
| "variant_id": "easy-click-checkboxes", | |
| "seed": 24, | |
| "max_steps": 18, | |
| }, | |
| {"reason": "low_progress_abort", "count": 2}, | |
| ) | |
| self.assertEqual(mutated["max_steps"], 20) | |
| def test_adaptive_step_budget_generator_respects_bounded_cap_across_repeated_mutations(self): | |
| generator = agents.AdaptiveStepBudgetGenerator() | |
| variant = { | |
| "task_id": "browsergym/miniwob.click-checkboxes", | |
| "task_family": "click", | |
| "difficulty": "medium", | |
| "variant_id": "easy-click-checkboxes", | |
| "seed": 24, | |
| "max_steps": 24, | |
| "base_max_steps": 24, | |
| } | |
| for _ in range(10): | |
| variant = generator.mutate_task(variant, {"reason": "max_steps_exceeded", "count": 3}) | |
| self.assertEqual(variant["max_steps"], 32) | |
| self.assertEqual(variant["step_budget_cap"], 32) | |
| self.assertGreaterEqual(variant["step_budget_adaptation_count"], 1) | |
| def test_judge_compact_trajectory_does_not_leak_env_verdict_columns(self): | |
| compact = agents._judge_compact_trajectory( | |
| [ | |
| { | |
| "step": 1, | |
| "observation": {"instruction": "Click Submit"}, | |
| "action": {"action_type": "click"}, | |
| "executed_action": {"action_type": "click", "target_id": "submit"}, | |
| "browsergym_action": "click('submit')", | |
| "browsergym_reward": 0.0, | |
| "reward_breakdown": {"total": 9.0}, | |
| "success": True, | |
| "failure_reason": "success", | |
| "constraints": {"oracle_calls": 0}, | |
| } | |
| ], | |
| 4, | |
| ) | |
| self.assertEqual(compact[0]["instruction"], "Click Submit") | |
| self.assertNotIn("reward_breakdown", compact[0]) | |
| self.assertNotIn("success", compact[0]) | |
| self.assertNotIn("failure_reason", compact[0]) | |
| self.assertNotIn("constraints", compact[0]) | |
| def test_environment_progress_evidence_detects_text_and_submit_progress(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| env = BrowserEnvironment() | |
| env.history = [{"action_type": "click", "target_id": "old-btn"}] | |
| previous_elements = [ | |
| BrowserElement( | |
| id="email", | |
| role="textbox", | |
| tag="input", | |
| text="", | |
| visible=True, | |
| enabled=True, | |
| attributes={"value": ""}, | |
| ), | |
| BrowserElement( | |
| id="cancel", | |
| role="button", | |
| tag="button", | |
| text="Cancel", | |
| visible=True, | |
| enabled=True, | |
| attributes={"semantic_hints": ["clickable"]}, | |
| ), | |
| ] | |
| current_elements = [ | |
| BrowserElement( | |
| id="email", | |
| role="textbox", | |
| tag="input", | |
| text="alice@example.com", | |
| visible=True, | |
| enabled=True, | |
| attributes={"value": "alice@example.com"}, | |
| ), | |
| BrowserElement( | |
| id="submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| visible=True, | |
| enabled=True, | |
| attributes={"semantic_hints": ["clickable", "submit_like"]}, | |
| ), | |
| ] | |
| env._current_elements_with_stats = lambda: ( # type: ignore[method-assign] | |
| current_elements, | |
| {"elements_after": len(current_elements)}, | |
| ) | |
| progress = env._assess_progress_evidence( | |
| previous_elements=previous_elements, | |
| action=BrowserAction(action_type="type", target_id="email", text="alice@example.com"), | |
| browser_reward=0.0, | |
| success=False, | |
| ) | |
| self.assertTrue(progress["progress_signal"]) | |
| self.assertEqual(progress["reward_credit"], 1) | |
| self.assertIn("text_entered", progress["signals"]) | |
| self.assertIn("submit_target_appeared", progress["signals"]) | |
| def test_find_clickable_submit_target_requires_visible_and_enabled(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| env = BrowserEnvironment() | |
| elements = [ | |
| BrowserElement( | |
| id="hidden-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| visible=False, | |
| enabled=True, | |
| attributes={"clickable": True}, | |
| ), | |
| BrowserElement( | |
| id="disabled-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| visible=True, | |
| enabled=False, | |
| attributes={"clickable": True}, | |
| ), | |
| BrowserElement( | |
| id="active-submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| visible=True, | |
| enabled=True, | |
| attributes={"clickable": True}, | |
| ), | |
| ] | |
| self.assertEqual(env._find_clickable_submit_target(elements), "active-submit") | |
| def test_environment_progress_evidence_detects_click_target_state_change(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| env = BrowserEnvironment() | |
| env.history = [{"action_type": "click", "target_id": "old-btn"}] | |
| previous_elements = [ | |
| BrowserElement( | |
| id="menu", | |
| role="button", | |
| tag="button", | |
| text="Menu", | |
| visible=True, | |
| enabled=True, | |
| attributes={"aria-expanded": "false", "class": "menu-toggle"}, | |
| ), | |
| ] | |
| current_elements = [ | |
| BrowserElement( | |
| id="menu", | |
| role="button", | |
| tag="button", | |
| text="Menu", | |
| visible=True, | |
| enabled=True, | |
| attributes={"aria-expanded": "true", "class": "menu-toggle open"}, | |
| ), | |
| ] | |
| env._current_elements_with_stats = lambda: ( # type: ignore[method-assign] | |
| current_elements, | |
| {"elements_after": len(current_elements)}, | |
| ) | |
| progress = env._assess_progress_evidence( | |
| previous_elements=previous_elements, | |
| action=BrowserAction(action_type="click", target_id="menu"), | |
| browser_reward=0.0, | |
| success=False, | |
| ) | |
| self.assertTrue(progress["progress_signal"]) | |
| self.assertEqual(progress["reward_credit"], 1) | |
| self.assertIn("click_target_state_changed", progress["signals"]) | |
| def test_environment_scales_no_progress_cap_with_variant_steps(self): | |
| with patch.dict( | |
| os.environ, | |
| { | |
| "BROWSER_ENV_DISABLE_LLM": "1", | |
| "BROWSER_ENV_MAX_NO_PROGRESS_STEPS": "4", | |
| "BROWSER_ENV_MAX_NO_PROGRESS_STEPS_CAP": "8", | |
| "BROWSER_ENV_NO_PROGRESS_STEP_RATIO": "0.5", | |
| }, | |
| clear=False, | |
| ): | |
| env = BrowserEnvironment() | |
| with patch.object(env.runtime, "reset", return_value=({}, {})): | |
| env.reset(task_id="browsergym/miniwob.click-test", episode_id="ep-easy") | |
| easy_cap = env.max_no_progress_steps | |
| env.reset(task_id="browsergym/miniwob.book-flight", episode_id="ep-hard") | |
| hard_cap = env.max_no_progress_steps | |
| self.assertEqual(easy_cap, 4) | |
| self.assertEqual(hard_cap, 8) | |
| def test_build_failure_detail_uses_task_failed_for_non_submit_terminal_failures(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| env = BrowserEnvironment() | |
| env.last_progress_evidence = {"progress_score": 0.0, "signals": []} | |
| detail = env._build_failure_detail( | |
| action=BrowserAction(action_type="click", target_id="target-1"), | |
| translated="click('target-1')", | |
| browser_reward=0.0, | |
| terminated=True, | |
| truncated=False, | |
| submit_like_failure=False, | |
| mode="task_failed", | |
| ) | |
| self.assertEqual(detail["mode"], "task_failed_terminal") | |
| def test_build_trackio_context_adds_project_filtered_dashboard_url(self): | |
| with patch.dict(os.environ, {}, clear=True): | |
| context = trainer.build_trackio_context( | |
| project="browser-rl-openenv", | |
| run_name="run-42", | |
| space_id="creovateHQ/browser-rl-trackio", | |
| enabled=True, | |
| available=True, | |
| ) | |
| self.assertTrue(context["configured"]) | |
| self.assertEqual(context["dashboard_url"], "https://creovatehq-browser-rl-trackio.hf.space") | |
| self.assertIn("project=browser-rl-openenv", context["project_dashboard_url"]) | |
| self.assertEqual(context["link_url"], context["project_dashboard_url"]) | |
| self.assertEqual( | |
| context["space_page_url"], | |
| "https://huggingface.co/spaces/creovateHQ/browser-rl-trackio", | |
| ) | |
| def test_make_env_wraps_remote_browser_client_in_sync_wrapper(self): | |
| env = trainer.make_env("https://example-space.hf.space") | |
| self.assertEqual(env.__class__.__name__, "SyncEnvClient") | |
| self.assertEqual(env.async_client.__class__.__name__, "BrowserEnv") | |
| env.close() | |
| def test_run_episode_prefers_executed_action_from_metadata(self): | |
| reset_obs = BrowserObservation( | |
| episode_id="ep-1", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click the button.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={}, | |
| ) | |
| next_obs = BrowserObservation( | |
| episode_id="ep-1", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click the button.", | |
| url="", | |
| step_index=1, | |
| max_steps=4, | |
| elements=[], | |
| history=[], | |
| constraints=ConstraintState(oracle_calls=1), | |
| reward_breakdown=RewardBreakdown(total=1.0), | |
| done=True, | |
| reward=1.0, | |
| success=True, | |
| failure_reason="success", | |
| metadata={"executed_action": {"action_type": "click", "target_id": "13"}}, | |
| ) | |
| class DummyEnv: | |
| def reset(self): | |
| return reset_obs | |
| def step(self, action): | |
| self.action = action | |
| return next_obs | |
| def close(self): | |
| return None | |
| summary = trainer.run_episode(DummyEnv(), "oracle") | |
| self.assertEqual(summary["trajectory"][0]["action"]["action_type"], "ask_oracle") | |
| self.assertEqual(summary["trajectory"][0]["executed_action"]["action_type"], "click") | |
| self.assertEqual(summary["trajectory"][0]["executed_action"]["target_id"], "13") | |
| def test_penalized_invalid_observation_only_uses_streak_reason_at_invalid_cap(self): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=False): | |
| env = BrowserEnvironment() | |
| env._state.step_count = env.current_variant.max_steps - 1 | |
| obs = env._penalized_invalid_observation( | |
| BrowserAction(action_type="click", target_id="missing"), | |
| "invalid_action", | |
| ) | |
| self.assertEqual(obs.failure_reason, "invalid_action") | |
| def test_extract_executed_action_recovers_remote_history_when_metadata_missing(self): | |
| next_obs = BrowserObservation( | |
| episode_id="ep-2", | |
| task_id="browsergym/miniwob.enter-text", | |
| task_family="form", | |
| difficulty="medium", | |
| instruction='Enter "Cristin" into the text field and press Submit.', | |
| url="", | |
| step_index=1, | |
| max_steps=8, | |
| elements=[], | |
| history=[ | |
| { | |
| "action_type": "type", | |
| "target_id": "14", | |
| "browsergym_action": "fill('14', 'Cristin')", | |
| "reward": -1.1, | |
| "status": "none", | |
| } | |
| ], | |
| constraints=ConstraintState(oracle_calls=1), | |
| reward_breakdown=RewardBreakdown(total=-1.1), | |
| done=False, | |
| reward=-1.1, | |
| success=False, | |
| failure_reason="none", | |
| metadata={}, | |
| ) | |
| recovered = trainer.extract_executed_action( | |
| next_obs, | |
| trainer.BrowserAction(action_type="ask_oracle", confidence=0.0), | |
| ) | |
| self.assertIsNotNone(recovered) | |
| self.assertEqual(recovered["action_type"], "type") | |
| self.assertEqual(recovered["target_id"], "14") | |
| self.assertEqual(recovered["text"], "Cristin") | |
| def test_replay_to_sft_rows_skips_raw_ask_oracle_steps(self): | |
| with TemporaryDirectory() as tmpdir: | |
| replay_path = Path(tmpdir) / "trajectories.jsonl" | |
| replay_path.write_text( | |
| "\n".join( | |
| [ | |
| '{"trajectory":[{"observation":{"instruction":"Click the button."},"action":{"action_type":"ask_oracle"},"executed_action":{"action_type":"click","target_id":"13"}}]}', | |
| '{"trajectory":[{"observation":{"instruction":"Click the button."},"action":{"action_type":"ask_oracle"},"executed_action":null}]}', | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| rows = slm_policy.replay_to_sft_rows([replay_path]) | |
| self.assertEqual(len(rows), 1) | |
| self.assertIn('"action_type": "click"', rows[0]["completion"]) | |
| def test_replay_to_sft_rows_can_filter_failed_noop_recovery(self): | |
| with TemporaryDirectory() as tmpdir: | |
| replay_path = Path(tmpdir) / "trajectories.jsonl" | |
| replay_path.write_text( | |
| "\n".join( | |
| [ | |
| '{"success":false,"trajectory":[{"observation":{"instruction":"Pick item."},"action_source":"noop_recovery","executed_action":{"action_type":"noop"}}]}', | |
| '{"success":true,"trajectory":[{"observation":{"instruction":"Click."},"action_source":"teacher_oracle","executed_action":{"action_type":"click","target_id":"13"}}]}', | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| rows = slm_policy.replay_to_sft_rows( | |
| [replay_path], | |
| success_only=True, | |
| allowed_action_sources={"teacher_oracle"}, | |
| ) | |
| self.assertEqual(len(rows), 1) | |
| self.assertIn('"action_type": "click"', rows[0]["completion"]) | |
| self.assertNotIn('"action_type": "noop"', rows[0]["completion"]) | |
| def test_parse_action_json_allows_unknown_target_id_when_not_strict(self): | |
| obs = BrowserObservation( | |
| episode_id="ep-3", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click submit.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[ | |
| BrowserElement( | |
| id="known", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True}, | |
| ) | |
| ], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={}, | |
| ) | |
| action = slm_policy.parse_action_json('{"action_type":"click","target_id":"unknown"}', obs, strict=False) | |
| self.assertEqual(action.target_id, "unknown") | |
| with self.assertRaises(ValueError): | |
| slm_policy.parse_action_json('{"action_type":"click","target_id":"unknown"}', obs, strict=True) | |
| def test_parse_action_json_repairs_partial_json_output(self): | |
| obs = BrowserObservation( | |
| episode_id="ep-3b", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click submit.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[ | |
| BrowserElement( | |
| id="known", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True}, | |
| ) | |
| ], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={}, | |
| ) | |
| action = slm_policy.parse_action_json( | |
| '{"action_type":"click","target_id":"known","confidence":0.8,"reasoning":"trunc', | |
| obs, | |
| ) | |
| self.assertEqual(action.action_type, "click") | |
| self.assertEqual(action.target_id, "known") | |
| self.assertAlmostEqual(action.confidence or 0.0, 0.8) | |
| self.assertEqual(action.reasoning, "partial_json_repair") | |
| def test_parse_action_json_recovers_browsergym_fill_action(self): | |
| obs = BrowserObservation( | |
| episode_id="ep-3c", | |
| task_id="browsergym/miniwob.choose-list", | |
| task_family="select", | |
| difficulty="easy", | |
| instruction="Choose Arielle.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[ | |
| BrowserElement( | |
| id="option-14", | |
| role="combobox", | |
| tag="select", | |
| text="", | |
| enabled=True, | |
| visible=True, | |
| attributes={"clickable": True}, | |
| ) | |
| ], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={}, | |
| ) | |
| action = slm_policy.parse_action_json("select_option('option-14', 'Arielle')", obs) | |
| self.assertEqual(action.action_type, "select") | |
| self.assertEqual(action.target_id, "option-14") | |
| self.assertEqual(action.text, "Arielle") | |
| self.assertEqual(action.reasoning, "browsergym_action_repair") | |
| def test_observation_to_prompt_includes_ranking_hints(self): | |
| obs = BrowserObservation( | |
| episode_id="ep-4", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click submit.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[ | |
| BrowserElement( | |
| id="submit", | |
| role="button", | |
| tag="button", | |
| text="Submit", | |
| enabled=True, | |
| visible=True, | |
| attributes={"semantic_hints": ["clickable", "submit_like"], "rank_score": 4.4}, | |
| ) | |
| ], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={"observation_stats": {"elements_before": 10, "elements_after": 4, "keep_ratio": 0.4}}, | |
| ) | |
| prompt = slm_policy.observation_to_prompt(obs) | |
| self.assertIn('"semantic_hints": ["clickable", "submit_like"]', prompt) | |
| self.assertIn('"observation_stats"', prompt) | |
| def test_compact_obs_includes_observation_filter_summary(self): | |
| obs = BrowserObservation( | |
| episode_id="ep-5", | |
| task_id="browsergym/miniwob.click-test", | |
| task_family="click", | |
| difficulty="easy", | |
| instruction="Click submit.", | |
| url="", | |
| step_index=0, | |
| max_steps=4, | |
| elements=[], | |
| history=[], | |
| constraints=ConstraintState(), | |
| reward_breakdown=RewardBreakdown(), | |
| done=False, | |
| reward=0.0, | |
| success=False, | |
| failure_reason="none", | |
| metadata={ | |
| "observation_filter": "heuristic_ranker_v1", | |
| "observation_stats": { | |
| "elements_before": 11, | |
| "elements_after": 5, | |
| "keep_ratio": 0.4545, | |
| "estimated_prompt_chars": 320, | |
| }, | |
| }, | |
| ) | |
| compact = trainer.compact_obs(obs) | |
| self.assertEqual(compact["observation_filter"], "heuristic_ranker_v1") | |
| self.assertEqual(compact["observation_stats"]["elements_before"], 11) | |
| self.assertEqual(compact["observation_stats"]["estimated_prompt_chars"], 320) | |
| def test_load_or_generate_returns_readable_error_payload_without_artifact(self): | |
| with TemporaryDirectory() as tmpdir: | |
| missing_artifact = Path(tmpdir) / "training_metrics.json" | |
| with patch.object(demo, "ARTIFACT", missing_artifact): | |
| with patch.dict(os.environ, {"BROWSER_ENV_DISABLE_LLM": "1"}, clear=True): | |
| payload = demo.load_or_generate() | |
| self.assertEqual(payload["status"], "error") | |
| self.assertIn("MINIWOB_URL", payload["error"]) | |
| self.assertFalse(payload["environment_status"]["miniwob_ready"]) | |
| def test_render_summary_markdown_includes_before_after_metrics(self): | |
| summary = demo.render_summary_markdown( | |
| { | |
| "before": { | |
| "success_rate": 0.2, | |
| "avg_reward": -2.0, | |
| "avg_steps": 9.0, | |
| "avg_oracle_calls": 1.0, | |
| "avg_invalid_actions": 2.0, | |
| }, | |
| "after": { | |
| "success_rate": 0.8, | |
| "avg_reward": 6.0, | |
| "avg_steps": 4.0, | |
| "avg_oracle_calls": 0.2, | |
| "avg_invalid_actions": 0.5, | |
| }, | |
| "deltas": { | |
| "success_rate": 0.6, | |
| "avg_reward": 8.0, | |
| "avg_steps": -5.0, | |
| "avg_oracle_calls": -0.8, | |
| "avg_invalid_actions": -1.5, | |
| }, | |
| } | |
| ) | |
| self.assertIn("Success rate", summary) | |
| self.assertIn("Average reward", summary) | |
| self.assertIn("Oracle calls", summary) | |
| def test_build_command_center_payload_prefers_artifact_trackio_metadata(self): | |
| artifact = { | |
| "run_started_at": "2026-04-25T00:00:00+00:00", | |
| "policy_mode": "heuristic_reference", | |
| "before": {}, | |
| "after": {}, | |
| "deltas": {}, | |
| "training_events": [{"phase": "after_eval"}], | |
| "trackio": { | |
| "enabled": True, | |
| "project": "browser-rl-openenv", | |
| "run_name": "run-42", | |
| "space_id": "creovateHQ/browser-rl-trackio", | |
| "dashboard_url": "https://creovatehq-browser-rl-trackio.hf.space", | |
| }, | |
| } | |
| environment_status = { | |
| "miniwob_ready": True, | |
| "llm_services_enabled": False, | |
| "openai_api_key_present": False, | |
| "trackio_available": True, | |
| } | |
| with patch.object(demo, "load_saved_artifact", return_value=(artifact, "")): | |
| with patch.object(demo, "collect_environment_status", return_value=environment_status): | |
| with patch.object(demo, "ARTIFACT", Path("/tmp/training_metrics.json")): | |
| payload = demo.build_command_center_payload() | |
| trackio = payload["artifact"]["trackio"] | |
| self.assertTrue(trackio["enabled"]) | |
| self.assertEqual(trackio["project"], "browser-rl-openenv") | |
| self.assertEqual(trackio["run_name"], "run-42") | |
| self.assertIn("project=browser-rl-openenv", trackio["project_dashboard_url"]) | |
| self.assertTrue(payload["runtime"]["trackio_configured"]) | |
| def test_serialize_task_presets_prefers_demo_quartet_order(self): | |
| variants = [ | |
| curriculum.TaskVariant("browsergym/miniwob.book-flight", "multi_step", "hard", "hard-book-flight", seed=47, max_steps=30), | |
| curriculum.TaskVariant("browsergym/miniwob.click-test", "click", "easy", "easy-click-test", seed=11, max_steps=8), | |
| curriculum.TaskVariant("browsergym/miniwob.choose-list", "select", "medium", "medium-choose-list", seed=31, max_steps=18), | |
| curriculum.TaskVariant("browsergym/miniwob.enter-text", "form", "medium", "medium-enter-text", seed=23, max_steps=14), | |
| curriculum.TaskVariant("browsergym/miniwob.extra-task", "multi_step", "hard", "hard-extra", seed=99, max_steps=40), | |
| ] | |
| class DummyPool: | |
| def __init__(self): | |
| self.variants = variants | |
| with patch.object(demo, "CurriculumPool", DummyPool): | |
| presets = demo._serialize_task_presets(limit=4) | |
| self.assertEqual( | |
| [preset["task_id"] for preset in presets], | |
| [ | |
| "browsergym/miniwob.click-test", | |
| "browsergym/miniwob.enter-text", | |
| "browsergym/miniwob.choose-list", | |
| "browsergym/miniwob.book-flight", | |
| ], | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |