feat(scenario): wave-2 observe drivers — assert_dom, inspect_filter_options, navigate_tabs, wait_for_selector (AGSCN-FR-024)

backend/src/services/dashboard_testing/execution/providers/browser_admission.py

@@ -36,12 +36,15 @@ _READ_ONLY_ACTIONS = frozenset({
     "open_dashboard", "wait_for_state", "refresh", "apply_native_filter",
     "navigate_tab", "inspect_filter_state", "apply_table_filter", "extract_table",
     "scroll_to", "inspect_columns", "click", "select_rows", "download",
+    # Wave-2 observe drivers (038.5.0, AGSCN-FR-024):
+    "assert_dom", "inspect_filter_options", "navigate_tabs", "wait_for_selector",
 })
 _MUTATION_ACTIONS = frozenset({"row_edit", "bulk_edit"})
 # Round-2 read-only actions that carry typed inputs; validated at admission before any I/O.
 _TYPED_READONLY_ACTIONS = frozenset({
     "navigate_tab", "apply_table_filter", "extract_table", "scroll_to", "click",
     "select_rows", "download",
+    "assert_dom", "inspect_filter_options", "wait_for_selector",
 })
 _MAX_DOWNLOAD_BYTES = 25 * 1024 * 1024  # 25 MiB (T034 spec)
 

backend/src/services/dashboard_testing/execution/providers/browser_readonly_actions.py

@@ -41,6 +41,12 @@ from src.services.dashboard_testing.execution.providers.browser_readonly_flows_n
     inspect_filter_state_flow,
     navigate_tab_flow,
 )
+from src.services.dashboard_testing.execution.providers.browser_readonly_flows_observe import (
+    assert_dom_flow,
+    inspect_filter_options_flow,
+    navigate_tabs_flow,
+    wait_for_selector_flow,
+)
 from src.services.dashboard_testing.execution.providers.browser_readonly_limits import (
     _ALLOWED_ARTIFACT_TYPES,
     _ALLOWED_SCROLL_DIRECTIONS,
@@ -56,16 +62,20 @@ from src.services.dashboard_testing.execution.providers.browser_readonly_limits
 
 __all__ = [
     "apply_table_filter_flow",
+    "assert_dom_flow",
     "click_flow",
     "download_flow",
     "extract_table_flow",
     "inspect_columns_flow",
+    "inspect_filter_options_flow",
     "inspect_filter_state_flow",
     "navigate_tab_flow",
+    "navigate_tabs_flow",
     "run_readonly_action_flow",
     "scroll_to_flow",
     "select_rows_flow",
     "validate_readonly_action_input",
+    "wait_for_selector_flow",
 ]
 
 # #region ScenarioExecution.BrowserProvider.ReadOnlyActions.Validate [C:3] [TYPE Function] [SEMANTICS provider,browser,readonly,validate]
@@ -120,7 +130,41 @@ def validate_readonly_action_input(action: str, action_input: dict[str, Any]) ->
         artifact_type = action_input.get("artifact_type")
         if artifact_type is not None and str(artifact_type) not in _ALLOWED_ARTIFACT_TYPES:
             return "BROWSER_DOWNLOAD_INVALID"
-    # inspect_filter_state, inspect_columns: no typed input required.
+    elif action == "assert_dom":
+        selector = action_input.get("selector")
+        if not isinstance(selector, str) or not selector.strip() or len(selector) > 512:
+            return "BROWSER_ASSERT_DOM_INVALID"
+        text_present = action_input.get("text_present")
+        text_absent = action_input.get("text_absent")
+        for value in (text_present, text_absent):
+            if value is not None and (not isinstance(value, str) or len(value) > 500):
+                return "BROWSER_ASSERT_DOM_INVALID"
+        for key in ("element_count", "min_count", "max_count"):
+            value = action_input.get(key)
+            if value is not None and (not isinstance(value, int) or isinstance(value, bool) or value < 0):
+                return "BROWSER_ASSERT_DOM_INVALID"
+    elif action == "inspect_filter_options":
+        for key in ("filter_id", "filter_name", "column"):
+            value = action_input.get(key)
+            if value is not None and (not isinstance(value, str) or not value.strip() or len(value) > 128):
+                return "BROWSER_FILTER_OPTIONS_INVALID"
+        limit = action_input.get("limit")
+        if limit is not None and (not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 500):
+            return "BROWSER_FILTER_OPTIONS_INVALID"
+    elif action == "wait_for_selector":
+        selector = action_input.get("selector")
+        text = action_input.get("text")
+        if (not isinstance(selector, str) or not selector.strip() or len(selector) > 512) and (
+            not isinstance(text, str) or not text.strip()
+        ):
+            return "BROWSER_WAIT_SELECTOR_INVALID"
+        state = action_input.get("state")
+        if state is not None and str(state) not in ("visible", "attached", "hidden"):
+            return "BROWSER_WAIT_STATE_INVALID"
+        timeout_ms = action_input.get("timeout_ms")
+        if timeout_ms is not None and (not isinstance(timeout_ms, int) or isinstance(timeout_ms, bool) or not 1 <= timeout_ms <= 120_000):
+            return "BROWSER_WAIT_SELECTOR_INVALID"
+    # inspect_filter_state, inspect_columns, navigate_tabs: no typed input required.
     return None
 # #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.Validate
 
@@ -155,6 +199,14 @@ async def run_readonly_action_flow(
         return await select_rows_flow(service, page, action_input, timeout_seconds=timeout_seconds), None
     if action == "download":
         return await download_flow(service, page, action_input, timeout_seconds=timeout_seconds)
+    if action == "assert_dom":
+        return await assert_dom_flow(service, page, action_input, timeout_seconds=timeout_seconds), None
+    if action == "inspect_filter_options":
+        return await inspect_filter_options_flow(service, page, action_input, timeout_seconds=timeout_seconds), None
+    if action == "navigate_tabs":
+        return await navigate_tabs_flow(service, page, action_input, timeout_seconds=timeout_seconds), None
+    if action == "wait_for_selector":
+        return await wait_for_selector_flow(service, page, action_input, timeout_seconds=timeout_seconds), None
     raise ValueError("BROWSER_ACTION_NOT_SUPPORTED")
 # #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.Dispatch
 # #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions

backend/src/services/dashboard_testing/execution/providers/browser_readonly_flows_observe.py

@@ -0,0 +1,235 @@
+# #region ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows [C:5] [TYPE Module] [SEMANTICS scenario,execution,provider,browser,observe,assert-dom,filter-options,wait-selector]
+# @ingroup ScenarioExecution
+# @BRIEF Wave-2 read-only observation drivers (038.5.0 AGSCN-FR-024): assert_dom DOM predicates,
+#        inspect_filter_options bounded dropdown options, wait_for_selector bounded wait,
+#        navigate_tabs composite tab sweep. Pure DOM reads plus tab clicks; no mutation.
+# @RELATION IMPLEMENTS -> [ScenarioExecution.BrowserProvider.Transport]
+# @RELATION DEPENDS_ON -> [ScenarioExecution.BrowserProvider.ReadOnlyActions.Limits]
+# @RELATION DEPENDS_ON -> [Plugin.Service.ScreenshotService]
+# @PRE The page holds an authenticated Superset dashboard session opened by _launch_and_login;
+#      inputs are admission-validated (browser_admission validate_readonly_action_input).
+# @POST Each flow returns typed details; a locator miss or violated predicate raises
+#       BrowserTransportSelectorNotFound / ValueError(BROWSER_ASSERT_FAILED) BEFORE evidence;
+#       unknown wait state raises BROWSER_WAIT_STATE_INVALID.
+# @INVARIANT Read-only: DOM reads, bounded waits and tab clicks only; no SQL, no API writes,
+#            no retries of locator misses; counts/bounds are capped by module limits.
+# @RATIONALE D-класс ассертов (038 Design Amendment 2026-09-17): DOM-проверки детерминированы
+#            и заменяют VLM/human там, где ожидание — собственный вход сценария (селектор/текст/
+#            счётчик/sticky), никогда встроенная бизнес-истина (EMBEDDED_METRIC_LITERAL guard
+#            в step_inputs.py).
+# @REJECTED Повторные попытки locator-miss и авто-retry нарушенных предикатов отвергнуты —
+#           fail-closed семантика транспорта (inconclusive, не тихий pass/fail).
+from __future__ import annotations
+
+from typing import Any
+
+from src.core.logger import logger
+
+from src.services.dashboard_testing.execution.providers.browser_native_filter import (
+    _ATTRIBUTE_IDENTITY_RE,
+    _DROPDOWN_MOUNT_TIMEOUT_MS,
+    _FILTER_BAR_SELECTORS,
+    _FILTER_CONTROL_SELECTORS,
+    BrowserTransportSelectorNotFound,
+)
+
+_SRC = "ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows"
+_MAX_FILTER_OPTIONS = 500
+_MAX_TAB_SWEEP = 25
+_MAX_SELECTOR_LENGTH = 512
+_MAX_TEXT_LENGTH = 500
+_ALLOWED_WAIT_SELECTOR_STATES = frozenset({"visible", "attached", "hidden"})
+_STICKY_HINTS = ("sticky", "fixed")
+# Generic antd tab-nav shapes for the composite sweep (legacy capture_dashboard_chunks DOM facts);
+# the per-{tab} templates in flows_nav remain owned by the singular navigate_tab action.
+_TAB_SWEEP_SELECTORS = (
+    ".ant-tabs-nav-list .ant-tabs-tab",
+    '[role="tablist"] [role="tab"]',
+)
+
+
+# #region ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.AssertDom [C:4] [TYPE Function] [SEMANTICS provider,browser,assert,dom,predicate]
+# @ingroup ScenarioExecution
+# @BRIEF assert_dom: evaluate typed DOM predicates against one resolved selector scope.
+# @POST Returns {selector, checks: {...}, passed}; a violated predicate raises
+#       ValueError("BROWSER_ASSERT_FAILED") (fail-closed before evidence); a missing selector
+#       raises BrowserTransportSelectorNotFound.
+async def assert_dom_flow(
+    service: Any, page: Any, action_input: dict[str, Any], *, timeout_seconds: float,
+) -> dict[str, Any]:
+    selector = str(action_input.get("selector") or "").strip()
+    if not selector or len(selector) > _MAX_SELECTOR_LENGTH:
+        raise ValueError("BROWSER_ASSERT_DOM_INVALID")
+    locator = await service._find_first_visible_locator([page.locator(selector)])
+    if locator is None:
+        logger.explore("assert_dom selector not found", src=_SRC, payload={"selector": selector}, error_code="BROWSER_SELECTOR_NOT_FOUND")
+        raise BrowserTransportSelectorNotFound("BROWSER_SELECTOR_NOT_FOUND")
+    checks: dict[str, Any] = {}
+    violations: list[str] = []
+
+    text_present = action_input.get("text_present")
+    if isinstance(text_present, str) and text_present.strip():
+        needle = text_present.strip()[:_MAX_TEXT_LENGTH]
+        body_text = str(await page.evaluate("() => document.body.innerText") or "")
+        found = needle in body_text
+        checks["text_present"] = needle
+        if not found:
+            violations.append(f"text_present:{needle}")
+
+    text_absent = action_input.get("text_absent")
+    if isinstance(text_absent, str) and text_absent.strip():
+        needle = text_absent.strip()[:_MAX_TEXT_LENGTH]
+        body_text = str(await page.evaluate("() => document.body.innerText") or "")
+        found = needle in body_text
+        checks["text_absent"] = needle
+        if found:
+            violations.append(f"text_absent:{needle}")
+
+    expected_href = action_input.get("href")
+    if isinstance(expected_href, str) and expected_href.strip():
+        actual = await page.evaluate("(el) => el.getAttribute('href')", locator.first)
+        checks["href"] = {"expected": expected_href.strip()[:2000], "actual": actual}
+        if actual != expected_href.strip():
+            violations.append("href")
+
+    element_count = action_input.get("element_count")
+    if isinstance(element_count, int) and not isinstance(element_count, bool):
+        count = min(await locator.count(), 1_000_000)
+        checks["element_count"] = {"expected": element_count, "actual": count}
+        if count != element_count:
+            violations.append("element_count")
+
+    min_count = action_input.get("min_count")
+    max_count = action_input.get("max_count")
+    if isinstance(min_count, int) or isinstance(max_count, int):
+        count = min(await locator.count(), 1_000_000)
+        checks["count_bounds"] = {"min": min_count, "max": max_count, "actual": count}
+        if isinstance(min_count, int) and count < min_count:
+            violations.append("min_count")
+        if isinstance(max_count, int) and count > max_count:
+            violations.append("max_count")
+
+    if action_input.get("sticky") is True:
+        position = str(await page.evaluate("(el) => getComputedStyle(el).position", locator.first) or "")
+        checks["sticky"] = {"position": position}
+        if position not in _STICKY_HINTS:
+            violations.append("sticky")
+
+    bounding_box = action_input.get("bounding_box")
+    if isinstance(bounding_box, dict) and bounding_box:
+        box = await page.evaluate("(el) => { const r = el.getBoundingClientRect(); return {x: r.x, y: r.y, width: r.width, height: r.height}; }", locator.first)
+        checks["bounding_box"] = {"expected": bounding_box, "actual": box}
+        for key in ("max_height_px",):
+            bound = bounding_box.get(key)
+            if isinstance(bound, (int, float)) and isinstance(box, dict) and float(box.get("height") or 0) > float(bound):
+                violations.append(f"bounding_box:{key}")
+
+    logger.reflect(
+        "assert_dom evaluated", src=_SRC,
+        payload={"selector": selector, "checks": len(checks), "violations": violations},
+    )
+    if violations:
+        raise ValueError("BROWSER_ASSERT_FAILED")
+    return {"selector": selector, "checks": checks, "passed": True}
+# #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.AssertDom
+
+
+# #region ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.FilterOptions [C:3] [TYPE Function] [SEMANTICS provider,browser,filter,options,observe]
+# @ingroup ScenarioExecution
+# @BRIEF inspect_filter_options: open the filter dropdown and read the bounded option list
+#        (B02/чеклист №4: опции фильтра «Выбранная метрика» per-tab). Read-only; no selection.
+# @POST Returns {filter_target, options: [...], count}; selector miss is typed, no retry.
+async def inspect_filter_options_flow(
+    service: Any, page: Any, action_input: dict[str, Any], *, timeout_seconds: float,
+) -> dict[str, Any]:
+    timeout_ms = int(timeout_seconds * 1000)
+    bar = await service._find_first_visible_locator([page.locator(s) for s in _FILTER_BAR_SELECTORS])
+    if bar is None:
+        logger.explore("Filter bar not found for options inspection", src=_SRC, error_code="BROWSER_SELECTOR_NOT_FOUND")
+        raise BrowserTransportSelectorNotFound("BROWSER_SELECTOR_NOT_FOUND")
+    control = await service._find_first_visible_locator([
+        bar.get_by_text(str(action_input.get("filter_id") or action_input.get("filter_name") or action_input.get("column") or ""), exact=False),
+        *[bar.locator(s) for s in _FILTER_CONTROL_SELECTORS],
+    ])
+    if control is None:
+        logger.explore("Filter control not found for options inspection", src=_SRC, error_code="BROWSER_SELECTOR_NOT_FOUND")
+        raise BrowserTransportSelectorNotFound("BROWSER_SELECTOR_NOT_FOUND")
+    await control.click(timeout=timeout_ms)
+    option_locator = page.locator(".ant-select-item-option, [role=option]")
+    try:
+        await option_locator.first.wait_for(state="visible", timeout=min(timeout_ms, _DROPDOWN_MOUNT_TIMEOUT_MS))
+    except Exception:
+        pass
+    total = min(await option_locator.count(), _MAX_FILTER_OPTIONS)
+    options: list[str] = []
+    for index in range(total):
+        label = str(await option_locator.nth(index).text_content() or "").strip()
+        if label:
+            options.append(label)
+    await page.keyboard.press("Escape")
+    return {
+        "filter_target": action_input.get("filter_id") or action_input.get("filter_name") or action_input.get("column"),
+        "options": options,
+        "count": len(options),
+    }
+# #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.FilterOptions
+
+
+# #region ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.WaitForSelector [C:2] [TYPE Function] [SEMANTICS provider,browser,wait,selector,bounded]
+# @ingroup ScenarioExecution
+# @BRIEF wait_for_selector: bounded single wait for a selector or visible text; fail-closed.
+# @POST Returns {selector, state, waited_ms} or raises BrowserTransportSelectorNotFound;
+#       unknown state raises BROWSER_WAIT_STATE_INVALID.
+async def wait_for_selector_flow(
+    service: Any, page: Any, action_input: dict[str, Any], *, timeout_seconds: float,
+) -> dict[str, Any]:
+    selector = str(action_input.get("selector") or "").strip()
+    text = str(action_input.get("text") or "").strip()
+    state = str(action_input.get("state") or "visible")
+    timeout_ms = min(int(action_input.get("timeout_ms") or timeout_seconds * 1000), 120_000)
+    if state not in _ALLOWED_WAIT_SELECTOR_STATES:
+        raise ValueError("BROWSER_WAIT_STATE_INVALID")
+    target = page.locator(selector) if selector else page.get_by_text(text, exact=False)
+    try:
+        await target.first.wait_for(state=state, timeout=timeout_ms)
+    except Exception:
+        logger.explore("wait_for_selector target never reached state", src=_SRC, payload={"selector": selector or text, "state": state}, error_code="BROWSER_SELECTOR_NOT_FOUND")
+        raise BrowserTransportSelectorNotFound("BROWSER_SELECTOR_NOT_FOUND")
+    return {"selector": selector or text, "state": state, "waited_ms": timeout_ms}
+# #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.WaitForSelector
+
+
+# #region ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.NavigateTabs [C:4] [TYPE Function] [SEMANTICS provider,browser,navigate,tabs,sweep,composite]
+# @ingroup ScenarioExecution
+# @BRIEF navigate_tabs: composite sweep — click every visible dashboard tab (bounded), return the
+#        visited list. Per-tab evidence/checkpoint stay the caller's (transport) responsibility;
+#        this driver only performs the navigation sequence.
+# @POST Returns {tabs: [...], visited_count}; a tab locator miss raises typed, no retry.
+async def navigate_tabs_flow(
+    service: Any, page: Any, action_input: dict[str, Any], *, timeout_seconds: float,
+) -> dict[str, Any]:
+    timeout_ms = int(timeout_seconds * 1000)
+    visited: list[str] = []
+    # Generic tab-container selectors (2026-09-12 live DOM facts + legacy capture_dashboard_chunks
+    # walk): the antd tab nav is the stable superset-independent shape; per-{tab} templates from
+    # _TAB_SELECTORS stay owned by the singular navigate_tab action.
+    for selector in _TAB_SWEEP_SELECTORS:
+        try:
+            candidates = await page.locator(selector).all()
+        except Exception:
+            candidates = []
+        for tab in candidates[:_MAX_TAB_SWEEP]:
+            label = str(await tab.text_content() or "").strip()
+            if not label or label in visited:
+                continue
+            await tab.click(timeout=timeout_ms)
+            visited.append(label)
+        if visited:
+            break
+    if not visited:
+        logger.explore("No dashboard tabs found for sweep", src=_SRC, error_code="BROWSER_SELECTOR_NOT_FOUND")
+        raise BrowserTransportSelectorNotFound("BROWSER_SELECTOR_NOT_FOUND")
+    logger.reflect("Tab sweep complete", src=_SRC, payload={"visited": len(visited)})
+    return {"tabs": visited, "visited_count": len(visited)}
+# #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows.NavigateTabs
+# #endregion ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows

backend/src/services/dashboard_testing/execution/providers/browser_transport.py

@@ -55,6 +55,8 @@ _READ_ONLY_ACTIONS = frozenset({
     "open_dashboard", "wait_for_state", "refresh", "apply_native_filter",
     "navigate_tab", "inspect_filter_state", "apply_table_filter", "extract_table",
     "scroll_to", "inspect_columns", "click", "select_rows", "download",
+    # Wave-2 observe drivers (038.5.0, AGSCN-FR-024):
+    "assert_dom", "inspect_filter_options", "navigate_tabs", "wait_for_selector",
 })
 _MUTATION_ACTIONS = frozenset({"row_edit", "bulk_edit"})
 _ALLOWED_WAIT_STATES = frozenset({"load", "domcontentloaded", "networkidle"})
@@ -68,6 +70,11 @@ _ROUND2_CHECKPOINTS = {
     "click": "clicked",
     "select_rows": "rows_selected",
     "download": "downloaded",
+    # Wave-2 observe checkpoints (AGSCN-FR-024):
+    "assert_dom": "dom_asserted",
+    "inspect_filter_options": "filter_options_inspected",
+    "navigate_tabs": "tabs_swept",
+    "wait_for_selector": "selector_awaited",
 }
 
 

backend/src/services/dashboard_testing/scenario/templates/__init__.py

@@ -32,10 +32,11 @@ REGISTERED_ACTIONS: dict[str, dict[str, Any]] = {
     "apply_table_filter": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "table_filter"},
     "pagination": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "pagination", "disabled": "pending driver"},
     "navigate_dashboard": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "cross_dashboard", "disabled": "pending driver"},
-    "assert_dom": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser", "disabled": "pending driver"},
-    "inspect_filter_options": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser", "disabled": "pending driver"},
-    "navigate_tabs": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "browser", "disabled": "pending driver"},
-    "wait_for_selector": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser", "disabled": "pending driver"},
+    # Wave-2 observe drivers implemented (browser_readonly_flows_observe.py); AGSCN-FR-025.
+    "assert_dom": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser"},
+    "inspect_filter_options": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser"},
+    "navigate_tabs": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "browser"},
+    "wait_for_selector": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser"},
     "extract_table": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser"},
     "scroll_to": {"tool": "browser", "phase": "interact", "risk": "browser_interaction", "capability": "browser"},
     "inspect_columns": {"tool": "browser", "phase": "observe", "risk": "read", "capability": "browser"},

backend/tests/services/dashboard_testing/registry/test_browser_observe_flows.py

@@ -0,0 +1,229 @@
+# #region Test.ScenarioExecution.BrowserObserveFlows [C:3] [TYPE Module] [SEMANTICS testing,scenario,provider,browser,observe,assert-dom,filter-options,wait-selector,navigate-tabs]
+# @defgroup Test.ScenarioExecution Wave-2 observe driver tests — assert_dom, inspect_filter_options,
+# wait_for_selector, navigate_tabs against a typed FakePage double.
+# @LAYER Test
+# @RELATION BINDS_TO -> [ScenarioExecution.BrowserProvider.ReadOnlyActions.ObserveFlows]
+# @RATIONALE Wave-2 drivers must be proven with typed doubles before any live canary; every typed
+#            rejection (BROWSER_ASSERT_FAILED, selector miss, invalid state) is fail-closed.
+# @REJECTED Testing drivers only through live Playwright was rejected — unit doubles keep the
+#           failure matrix deterministic and independent of the stand.
+
+from __future__ import annotations
+
+import pytest
+
+from src.services.dashboard_testing.execution.providers.browser_native_filter import (
+    BrowserTransportSelectorNotFound,
+)
+from src.services.dashboard_testing.execution.providers.browser_readonly_flows_observe import (
+    assert_dom_flow,
+    inspect_filter_options_flow,
+    navigate_tabs_flow,
+    wait_for_selector_flow,
+)
+
+
+class FakeLocator:
+    def __init__(self, *, visible=True, count=1, text="", href=None, position=None, children=None, items=None):
+        self._visible = visible
+        self._count = count
+        self._text = text
+        self._href = href
+        self._position = position
+        self._children = children or {}
+        self._items = items or []
+
+    async def count(self) -> int:
+        return self._count
+
+    async def is_visible(self) -> bool:
+        return self._visible
+
+    async def text_content(self) -> str:
+        return self._text
+
+    async def click(self, timeout=None):
+        return None
+
+    async def all(self):
+        return self._items or [self]
+
+    @property
+    def first(self):
+        return self._items[0] if self._items else self
+
+    def nth(self, index):
+        return self._items[index] if self._items else self
+
+    def locator(self, selector: str) -> "FakeLocator":
+        return self._children.get(selector, FakeLocator(visible=False, count=0))
+
+    def get_by_text(self, text: str, exact: bool = False) -> "FakeLocator":
+        # Playwright locators expose get_by_text; the options flow probes the filter by label.
+        if self._visible and text in (self._text or ""):
+            return self
+        for item in self._items:
+            if text in (item._text or ""):
+                return item
+        for child in self._children.values():
+            if text in (child._text or ""):
+                return child
+        return FakeLocator(visible=False, count=0)
+
+    async def wait_for(self, state="visible", timeout=None):
+        if not self._visible and state == "visible":
+            raise RuntimeError("never visible")
+
+
+class FakePage:
+    def __init__(self, *, selectors=None, body_text="", options=None, tabs=None):
+        self._selectors = selectors or {}
+        self._body_text = body_text
+        self._options = options or []
+        self._tabs = tabs or []
+        self.url = "http://superset.test/superset/dashboard/11/"
+        self.escape_pressed = False
+
+    def locator(self, selector: str) -> FakeLocator:
+        if selector == ".ant-select-item-option, [role=option]":
+            option_items = [FakeLocator(text=option, visible=True) for option in self._options]
+            return FakeLocator(count=len(option_items), items=option_items, visible=bool(option_items))
+        return self._selectors.get(selector, FakeLocator(visible=False, count=0))
+
+    def get_by_text(self, text: str, exact: bool = False) -> FakeLocator:
+        return FakeLocator(visible=text in self._body_text or text in [t for t, _ in self._tabs], count=1)
+
+    class keyboard:
+        @staticmethod
+        async def press(key: str):
+            return None
+
+    async def evaluate(self, script: str, arg=None):
+        if "innerText" in script:
+            return self._body_text
+        if "href" in script:
+            return self._selectors.get("__href__", FakeLocator())._href
+        if "getBoundingClientRect" in script:
+            return {"x": 0, "y": 0, "width": 1920, "height": 120}
+        if "position" in script:
+            return "sticky"
+        raise AssertionError(f"unexpected evaluate script: {script}")
+
+
+class FakeService:
+    async def _find_first_visible_locator(self, candidates):
+        for locator in candidates:
+            try:
+                if await locator.count() > 0 and await locator.first.is_visible():
+                    return locator.first
+            except Exception:
+                continue
+        return None
+
+
+def _page_with_body(text: str) -> FakePage:
+    page = FakePage(body_text=text)
+    page._selectors["table th"] = FakeLocator(count=5, position="sticky", href=None)
+    return page
+
+
+@pytest.mark.asyncio()
+async def test_assert_dom_text_present_and_absent_pass():
+    service = FakeService()
+    page = _page_with_body("Нет данных\nНужна помощь?")
+    details = await assert_dom_flow(
+        service, page, {"selector": "table th", "text_present": "Нет данных", "text_absent": "DB engine"},
+        timeout_seconds=1.0,
+    )
+    assert details["passed"] is True
+    assert details["checks"]["text_present"] == "Нет данных"
+
+
+@pytest.mark.asyncio()
+async def test_assert_dom_text_absent_violation_fails_closed():
+    service = FakeService()
+    page = _page_with_body("DB engine error on chart 3")
+    with pytest.raises(ValueError, match="BROWSER_ASSERT_FAILED"):
+        await assert_dom_flow(
+            service, page, {"selector": "table th", "text_absent": "DB engine"},
+            timeout_seconds=1.0,
+        )
+
+
+@pytest.mark.asyncio()
+async def test_assert_dom_selector_miss_is_typed():
+    service = FakeService()
+    page = FakePage()
+    with pytest.raises(BrowserTransportSelectorNotFound):
+        await assert_dom_flow(service, page, {"selector": ".missing"}, timeout_seconds=1.0)
+
+
+@pytest.mark.asyncio()
+async def test_assert_dom_element_count_bounds():
+    service = FakeService()
+    page = _page_with_body("x")
+    details = await assert_dom_flow(
+        service, page, {"selector": "table th", "min_count": 1, "max_count": 10},
+        timeout_seconds=1.0,
+    )
+    assert details["checks"]["count_bounds"]["actual"] == 5
+    with pytest.raises(ValueError, match="BROWSER_ASSERT_FAILED"):
+        await assert_dom_flow(
+            service, page, {"selector": "table th", "element_count": 2},
+            timeout_seconds=1.0,
+        )
+
+
+@pytest.mark.asyncio()
+async def test_inspect_filter_options_returns_bounded_list():
+    service = FakeService()
+    page = FakePage(options=["Все метрики", "ПДЗ", "Оборот"])
+    # The flow resolves the bar via _FILTER_BAR_SELECTORS, then the control by label text.
+    from src.services.dashboard_testing.execution.providers.browser_native_filter import _FILTER_BAR_SELECTORS
+    page._selectors[_FILTER_BAR_SELECTORS[0]] = FakeLocator(count=1, visible=True, text="Выбранная метрика")
+    details = await inspect_filter_options_flow(
+        service, page, {"filter_name": "Выбранная метрика"}, timeout_seconds=1.0,
+    )
+    assert details["count"] == 3
+    assert details["options"][0] == "Все метрики"
+
+
+@pytest.mark.asyncio()
+async def test_wait_for_selector_visible_passes_and_missing_fails():
+    service = FakeService()
+    page = _page_with_body("Сводка")
+    details = await wait_for_selector_flow(
+        service, page, {"selector": "table th", "state": "visible"}, timeout_seconds=1.0,
+    )
+    assert details["state"] == "visible"
+    missing = FakePage()
+    with pytest.raises(BrowserTransportSelectorNotFound):
+        await wait_for_selector_flow(
+            service, missing, {"selector": ".never", "state": "visible"}, timeout_seconds=0.1,
+        )
+    with pytest.raises(ValueError, match="BROWSER_WAIT_STATE_INVALID"):
+        await wait_for_selector_flow(
+            service, page, {"selector": "table th", "state": "bogus"}, timeout_seconds=1.0,
+        )
+
+
+@pytest.mark.asyncio()
+async def test_navigate_tabs_sweeps_visible_tabs():
+    service = FakeService()
+    tab_items = [FakeLocator(text="ПДЗ (платёжный)"), FakeLocator(text="Оборот"), FakeLocator(text="Сводка")]
+    page = FakePage()
+    # The sweep walks the generic antd tab-nav shapes (_TAB_SWEEP_SELECTORS).
+    page._selectors[".ant-tabs-nav-list .ant-tabs-tab"] = FakeLocator(count=len(tab_items), items=tab_items)
+
+    details = await navigate_tabs_flow(service, page, {}, timeout_seconds=1.0)
+    assert details["visited_count"] == 3
+    assert details["tabs"] == ["ПДЗ (платёжный)", "Оборот", "Сводка"]
+
+
+@pytest.mark.asyncio()
+async def test_navigate_tabs_no_tabs_is_typed():
+    service = FakeService()
+    page = FakePage()
+    with pytest.raises(BrowserTransportSelectorNotFound):
+        await navigate_tabs_flow(service, page, {}, timeout_seconds=0.1)
+# #endregion Test.ScenarioExecution.BrowserObserveFlows

backend/tests/services/dashboard_testing/scenario/test_validator.py

@@ -116,10 +116,11 @@ def test_disabled_action_rejected_before_io() -> None:
     assert any("pagination" in f.message for f in result.errors)
 
 
-def test_disabled_new_wave_action_rejected() -> None:
-    # Wave-2 drivers (assert_dom etc.) are registered disabled until their canary gates close.
+def test_disabled_navigate_dashboard_rejected() -> None:
+    # Wave-2 observe drivers (assert_dom etc.) are implemented and enabled; navigate_dashboard
+    # stays disabled: pending driver until its own canary gate closes.
     data = _load_raw("scenario_valid.json")
-    data["steps"][0]["action"] = "assert_dom"
+    data["steps"][0]["action"] = "navigate_dashboard"
     data["steps"][0]["tool"] = "browser"
     result = validate_scenario(DashboardTestScenario.model_validate(data))
     assert result.valid is False