diff --git a/openadapt_flow/backends/playwright_backend.py b/openadapt_flow/backends/playwright_backend.py index 37acccee..58eb6bb1 100644 --- a/openadapt_flow/backends/playwright_backend.py +++ b/openadapt_flow/backends/playwright_backend.py @@ -180,6 +180,20 @@ class _FramePoint: if (cursor === row) break; cursor = parent; } + // Retain the complete live DOM state for the bounded record boundary, + // excluding only OpenAdapt's own random lease attribute. This remains in + // the page-local guard store; it is never returned in a report. Comparing + // the final state, rather than treating every MutationObserver callback as + // irreversible, admits framework-owned hover/actionability churn only + // when the exact same Element and DOM state have been restored. + const boundary = (row || node).cloneNode(true); + for (const candidate of [boundary, ...boundary.querySelectorAll('*')]) { + for (const attr of Array.from(candidate.attributes || [])) { + if (attr.name.startsWith('data-openadapt-actuation-')) { + candidate.removeAttribute(attr.name); + } + } + } return { descriptor: JSON.stringify([ 1, @@ -194,6 +208,7 @@ class _FramePoint: ], ancestry, rowIdentity, + boundary.outerHTML, ]), rowIdentity: rowIdentity, row: row, @@ -253,17 +268,6 @@ class _FramePoint: tokenMap.delete(args.token); }; entry.invalidate = invalidate; - // Any mutation in the target's record boundary after arming invalidates - // the lease. Descriptor equality is deliberately irrelevant: hidden - // attributes and pixel-identical node replacement can still change the - // action that a click performs. - entry.observer = new MutationObserver(invalidate); - entry.observer.observe(observed.row || el, { - attributes: true, - childList: true, - characterData: true, - subtree: true, - }); if (args.requireFocused) { if (document.activeElement !== el) { invalidate(); @@ -1361,12 +1365,15 @@ def act_structural( """Atomically verify and click the exact DOM target resolved earlier. A strict locator must still resolve to the token-bound Element and its - unchanged target/row descriptor. A short-lived MutationObserver removes - the token on intervening identity mutation. The final action is a - Playwright click/dblclick on that unique random-token locator, preserving - Playwright's native pointer sequence and actionability checks rather - than synthesizing DOM events. A replacement element or changed record - row is a refusal, never a coordinate fallback. + unchanged complete bounded DOM descriptor. The same checks are repeated + after Playwright's actionability trial and immediately before the real + input, so transient framework presentation churn is admitted only when + the exact same Element and DOM state have been restored. The final + action is a Playwright click/dblclick on that unique random-token + locator, preserving Playwright's native pointer sequence rather than + synthesizing DOM events. A replacement element, lasting hidden + attribute change, or changed record row is a refusal, never a coordinate + fallback. """ fingerprint = handle.target_fingerprint diff --git a/scripts/export_public_demo_evidence.py b/scripts/export_public_demo_evidence.py index 584e001f..1094be88 100644 --- a/scripts/export_public_demo_evidence.py +++ b/scripts/export_public_demo_evidence.py @@ -355,6 +355,55 @@ def _probe_frame_presentation_times_us( return presentation_times_us +def _probe_video_sample_end_us( + ffprobe: str, + media_path: Path, +) -> int: + """Return the exact end of the final encoded video sample. + + A frame timestamp identifies the start of a sample, not the media duration. + The public viewer must retain the complete terminal hold, so bind the + timeline to ``max(PTS + duration)`` across the encoded video packets. + """ + + result = subprocess.run( + [ + ffprobe, + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "packet=pts_time,duration_time", + "-of", + "json", + str(media_path), + ], + check=True, + capture_output=True, + text=True, + ) + payload = json.loads(result.stdout) + raw_packets = payload.get("packets") if isinstance(payload, dict) else None + if not isinstance(raw_packets, list) or not raw_packets: + raise EvidencePackError(f"ffprobe found no video packets in {media_path}") + sample_end_us = 0 + for item in raw_packets: + pts = item.get("pts_time") if isinstance(item, dict) else None + duration = item.get("duration_time") if isinstance(item, dict) else None + if not isinstance(pts, str) or not isinstance(duration, str): + raise EvidencePackError(f"ffprobe omitted packet timing in {media_path}") + end_us = (Decimal(pts) + Decimal(duration)) * Decimal(1_000_000) + if end_us != end_us.to_integral_value(): + raise EvidencePackError( + f"ffprobe returned a sub-microsecond sample end in {media_path}" + ) + sample_end_us = max(sample_end_us, int(end_us)) + if sample_end_us <= 0: + raise EvidencePackError(f"ffprobe returned an empty duration for {media_path}") + return sample_end_us + + def _write_presentation_clip( *, capture: _PresentationCapture, @@ -397,7 +446,7 @@ def _write_presentation_clip( ffmpeg, ffprobe = _presentation_tools() presentation_dir.mkdir(parents=True, exist_ok=True) - media_path = presentation_dir / f"{clip_id}.webm" + media_path = presentation_dir / f"{clip_id}.mp4" timeline_path = presentation_dir / f"{clip_id}.control-overlay.v2.json" pts_path = presentation_dir / f"{clip_id}.frame-pts-us.json" with tempfile.TemporaryDirectory( @@ -453,17 +502,23 @@ def _write_presentation_clip( str(concat_path), "-an", "-fps_mode", - "vfr", + "passthrough", + "-enc_time_base", + "demux", "-c:v", - "libvpx-vp9", - "-lossless", - "1", - "-deadline", - "good", - "-cpu-used", - "2", + "libx264", + "-preset", + "slow", + "-crf", + "18", + "-bf", + "0", "-pix_fmt", "yuv420p", + "-video_track_timescale", + "1000000", + "-movflags", + "+faststart", str(media_path), ], check=True, @@ -493,7 +548,16 @@ def _write_presentation_clip( ) media_sha256 = _sha256(media_path) - duration_ms = presentation_times_us[-1] // 1_000 + sample_end_us = _probe_video_sample_end_us(ffprobe, media_path) + if sample_end_us <= presentation_times_us[-1]: + raise EvidencePackError( + f"{clip_id} final decoded sample has no retained duration" + ) + if sample_end_us % 1_000: + raise EvidencePackError( + f"{clip_id} final decoded sample does not end on a media millisecond" + ) + duration_ms = sample_end_us // 1_000 timeline = build_runtime_control_overlay_timeline_v2( capture.frames, data_classification=ControlOverlayDataClassification.SYNTHETIC, @@ -1247,6 +1311,7 @@ def _media_type(path: Path) -> str: ".html": "text/html", ".py": "text/x-python", ".png": "image/png", + ".mp4": "video/mp4", ".webm": "video/webm", }.get(path.suffix.lower(), "application/octet-stream") @@ -1258,7 +1323,7 @@ def _role(path: str) -> str: return "compiled_bundle" if "/qualification/" in path: return "qualification" - if path.endswith(".webm"): + if path.endswith((".mp4", ".webm")): return "media" if "/cases/" in path: return "case_evidence" @@ -1345,7 +1410,7 @@ def _assemble_manifest( def presentation_media(clip_id: str) -> dict[str, Any]: return { - "media": _ref_for(root, presentation / f"{clip_id}.webm"), + "media": _ref_for(root, presentation / f"{clip_id}.mp4"), "timeline": _ref_for( root, presentation / f"{clip_id}.control-overlay.v2.json" ), @@ -1754,7 +1819,7 @@ def _validate_presentation_artifacts( pts_relative = str(clip["frame_pts"]["path"]) expected_prefix = f"artifacts/presentation/{clip_id}" if ( - media_relative != f"{expected_prefix}.webm" + media_relative not in {f"{expected_prefix}.mp4", f"{expected_prefix}.webm"} or timeline_relative != f"{expected_prefix}.control-overlay.v2.json" or pts_relative != f"{expected_prefix}.frame-pts-us.json" ): @@ -1806,13 +1871,22 @@ def _validate_presentation_artifacts( raise EvidencePackError( f"exact-PTS sidecar does not bind decoded media: {pts_relative}" ) + expected_duration_ms = presentation_times_us[-1] // 1_000 + if media_path.suffix.lower() == ".mp4": + _ffmpeg, ffprobe = _presentation_tools() + sample_end_us = _probe_video_sample_end_us(ffprobe, media_path) + if sample_end_us % 1_000: + raise EvidencePackError( + f"presentation media does not end on a millisecond: {media_relative}" + ) + expected_duration_ms = sample_end_us // 1_000 if ( timeline.get("schema_version") != "openadapt.control-overlay-timeline/v2" or timeline.get("data_classification") != "synthetic" or timeline.get("evidence_pack_id") != pack_id or timeline.get("media_sha256") != media_sha256 or timeline.get("media_frame_count") != len(presentation_times_us) - or timeline.get("duration_ms") != presentation_times_us[-1] // 1_000 + or timeline.get("duration_ms") != expected_duration_ms ): raise EvidencePackError( f"control-overlay timeline does not bind media: {timeline_relative}" diff --git a/scripts/openimis_eligibility_demo.py b/scripts/openimis_eligibility_demo.py index 440d9420..d1241f57 100644 --- a/scripts/openimis_eligibility_demo.py +++ b/scripts/openimis_eligibility_demo.py @@ -72,9 +72,8 @@ OpenIMISFixture, ) -from openadapt_flow.backend import StructuralResolutionRefused # noqa: E402 from openadapt_flow.backends.playwright_backend import PlaywrightBackend # noqa: E402 -from openadapt_flow.ir import StructuralHandle, StructuralLocator # noqa: E402 +from openadapt_flow.ir import StructuralLocator # noqa: E402 HERE = Path(__file__).resolve().parent DEPLOYMENT_YAML = ( @@ -99,7 +98,10 @@ class names/ids, and the service options in a portal outside the dialog. the checked policyholder. This narrow adapter records stable semantic locators and a structured identity string that includes the run's insuree number for dialog-scoped actions. The compiler parameterizes that value, - so replay verifies the current run's policyholder before acting. + so replay verifies the current run's policyholder before acting. Structural + delivery remains inherited from :class:`PlaywrightBackend`, including its + unique-candidate refusal and one-shot mutation fingerprint between resolve + and actuation. """ _STRUCTURAL_JS = r"""([px, py]) => { @@ -109,14 +111,14 @@ class names/ids, and the service options in a portal outside the dialog. const placeholder = input ? (input.getAttribute('placeholder') || '') : ''; if (placeholder.startsWith('Insuree enquiry')) { return { - selector: 'input[placeholder^="Insuree enquiry"]', + selector: 'input[placeholder^="Insuree enquiry"]:visible', role: 'textbox', name: 'Insuree enquiry' }; } if (placeholder.startsWith('Search Service')) { return { - selector: 'input[placeholder^="Search Service"]', + selector: 'input[placeholder^="Search Service"]:visible', role: 'textbox', name: 'Search Service' }; @@ -210,78 +212,6 @@ def structured_text_at(self, x: int, y: int) -> str | None: context["insurance_no"] = dialogs[0][0] return json.dumps(context, separators=(",", ":")) - def locate_structural(self, locator: StructuralLocator) -> StructuralHandle | None: - """Refuse duplicate semantic candidates instead of falling to pixels. - - The generic browser backend treats an ambiguous structural locator as - a miss so less specialized workflows may continue down the evidence - ladder. This reference adapter authors exact semantic locators for - every clicked control, so more than one live candidate is a record/ - dialog ambiguity and must halt rather than let a visual rung pick one. - """ - if locator.selector: - candidates = self.page.locator(locator.selector) - elif locator.role and locator.name: - candidates = self.page.get_by_role( - locator.role, # type: ignore[arg-type] - name=locator.name, - exact=True, - ) - else: - return None - try: - visible_indices = candidates.evaluate_all( - """elements => elements.flatMap((el, index) => { - const style = window.getComputedStyle(el); - const rect = el.getBoundingClientRect(); - const visible = style.display !== 'none' - && style.visibility !== 'hidden' - && style.visibility !== 'collapse' - && rect.width > 0 && rect.height > 0 - && rect.right > 0 && rect.bottom > 0 - && rect.left < window.innerWidth - && rect.top < window.innerHeight; - return visible ? [index] : []; - })""" - ) - if not isinstance(visible_indices, list) or not all( - isinstance(index, int) for index in visible_indices - ): - return None - if len(visible_indices) > 1: - raise StructuralResolutionRefused( - "openIMIS eligibility target is structurally ambiguous" - ) - if not visible_indices: - return None - candidate = candidates.nth(visible_indices[0]) - box = candidate.bounding_box() - if not box or box["width"] <= 0 or box["height"] <= 0: - return None - cx = int(round(box["x"] + box["width"] / 2)) - cy = int(round(box["y"] + box["height"] / 2)) - vw, vh = self.viewport - if not (0 <= cx < vw and 0 <= cy < vh): - return None - topmost = candidate.evaluate( - """(el, pt) => { - const n = document.elementFromPoint(pt[0], pt[1]); - return !!n && (n === el || el.contains(n)); - }""", - [cx, cy], - ) - if not topmost: - return None - return StructuralHandle(point=(cx, cy), candidate_count=1) - except StructuralResolutionRefused: - raise - except Exception: - # Observation failures are a miss; the resolver may use other - # evidence, but a proven multi-candidate ambiguity above is a - # terminal refusal and never reaches a weaker rung. - return None - - def eligibility_effects() -> list[Any]: """The eligibility outcome bound to the run's ``insurance_no``. diff --git a/tests/test_openimis_eligibility.py b/tests/test_openimis_eligibility.py index f2361c5c..d9ab17b3 100644 --- a/tests/test_openimis_eligibility.py +++ b/tests/test_openimis_eligibility.py @@ -18,7 +18,7 @@ import pytest -from openadapt_flow.backend import StructuralResolutionRefused +from openadapt_flow.backends.playwright_backend import PlaywrightBackend from openadapt_flow.deployment import build_effect_verifier, load_deployment from openadapt_flow.ir import StructuralLocator from openadapt_flow.runtime.effects import ( @@ -64,7 +64,7 @@ def evaluate(script: str, point: list[int]) -> object: "dialog_identifiers": [["999000003"]], } return { - "selector": 'input[placeholder^="Search Service"]', + "selector": 'input[placeholder^="Search Service"]:visible', "role": "textbox", "name": "Search Service", } @@ -72,7 +72,7 @@ def evaluate(script: str, point: list[int]) -> object: backend = OpenIMISEligibilityBackend(Page()) # type: ignore[arg-type] locator = backend.structural_locator_at(320, 240) assert locator == StructuralLocator( - selector='input[placeholder^="Search Service"]', + selector='input[placeholder^="Search Service"]:visible', role="textbox", name="Search Service", ) @@ -115,63 +115,11 @@ def evaluate(script: str, point: list[int]) -> object: assert backend.structured_text_at(320, 240) is None -def test_browser_adapter_refuses_ambiguous_structural_candidates() -> None: - class Candidates: - @staticmethod - def evaluate_all(script: str) -> list[int]: - return [0, 1] - - class Page: - @staticmethod - def locator(selector: str) -> Candidates: - assert selector == 'input[placeholder^="Search Service"]' - return Candidates() - - backend = OpenIMISEligibilityBackend(Page()) # type: ignore[arg-type] - with pytest.raises(StructuralResolutionRefused, match="ambiguous"): - backend.locate_structural( - StructuralLocator(selector='input[placeholder^="Search Service"]') - ) - - -def test_browser_adapter_ignores_hidden_structural_duplicate() -> None: - class Candidate: - @staticmethod - def bounding_box() -> dict[str, float]: - return {"x": 100, "y": 50, "width": 40, "height": 20} - - @staticmethod - def evaluate(script: str, point: list[int]) -> bool: - assert point == [120, 60] - return True - - class Candidates: - @staticmethod - def evaluate_all(script: str) -> list[int]: - # CSS matched a hidden React template at index 0 and the sole live - # control at index 1; only the latter is returned by the JS filter. - return [1] - - @staticmethod - def nth(index: int) -> Candidate: - assert index == 1 - return Candidate() - - class Page: - viewport_size = {"width": 1280, "height": 800} - - @staticmethod - def locator(selector: str) -> Candidates: - assert selector == 'input[placeholder^="Search Service"]' - return Candidates() - - backend = OpenIMISEligibilityBackend(Page()) # type: ignore[arg-type] - handle = backend.locate_structural( - StructuralLocator(selector='input[placeholder^="Search Service"]') +def test_browser_adapter_uses_the_hardened_playwright_actuation_guard() -> None: + assert ( + OpenIMISEligibilityBackend.locate_structural + is PlaywrightBackend.locate_structural ) - assert handle is not None - assert handle.point == (120, 60) - assert handle.candidate_count == 1 def _report(*, success: bool, verdict: str | None, earlier_ok: bool = True): diff --git a/tests/test_playwright_frame_actuation.py b/tests/test_playwright_frame_actuation.py index 04d0d526..e18303b7 100644 --- a/tests/test_playwright_frame_actuation.py +++ b/tests/test_playwright_frame_actuation.py @@ -79,6 +79,42 @@ def test_nested_frame_locator_delivers_once_in_top_level_coordinates( backend.act_structural(locator, handle) +def test_transient_actionability_attribute_churn_is_revalidated_and_admitted( + framed_backend, +) -> None: + backend, _page, frame = framed_backend + frame.locator("#target").evaluate( + """el => el.addEventListener('mousemove', () => { + el.setAttribute('data-focus', 'true'); + el.removeAttribute('data-focus'); + })""" + ) + locator, _box = _target_locator(backend, frame) + handle = backend.locate_structural(locator) + assert handle is not None + + backend.act_structural(locator, handle) + + assert frame.evaluate("window.clicks") == 1 + + +def test_lasting_hidden_target_attribute_change_refuses_delivery( + framed_backend, +) -> None: + backend, _page, frame = framed_backend + locator, _box = _target_locator(backend, frame) + handle = backend.locate_structural(locator) + assert handle is not None + + frame.locator("#target").evaluate( + "el => el.setAttribute('data-action-route', 'wrong-record')" + ) + + with pytest.raises(StructuralResolutionRefused, match="changed before delivery"): + backend.act_structural(locator, handle) + assert frame.evaluate("window.clicks") == 0 + + def test_frame_path_and_target_ambiguity_refuse_before_arming() -> None: sync = pytest.importorskip("playwright.sync_api") from openadapt_flow.backends.playwright_backend import PlaywrightBackend diff --git a/tests/test_public_demo_presentation_media.py b/tests/test_public_demo_presentation_media.py new file mode 100644 index 00000000..67970cb8 --- /dev/null +++ b/tests/test_public_demo_presentation_media.py @@ -0,0 +1,96 @@ +"""Characterization of exact presentation-media timing.""" + +from __future__ import annotations + +import io +import json +import shutil +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest +from PIL import Image + +from scripts.export_public_demo_evidence import ( + _PresentationCapture, + _probe_video_sample_end_us, + _write_presentation_clip, +) + + +def _png(color: tuple[int, int, int]) -> bytes: + output = io.BytesIO() + Image.new("RGB", (64, 64), color).save(output, format="PNG") + return output.getvalue() + + +def test_sample_end_includes_the_final_packet_duration( + monkeypatch, tmp_path: Path +) -> None: + payload = { + "packets": [ + {"pts_time": "0.000000", "duration_time": "0.250000"}, + {"pts_time": "0.250000", "duration_time": "1.750000"}, + {"pts_time": "2.000000", "duration_time": "0.001000"}, + ] + } + monkeypatch.setattr( + subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(stdout=json.dumps(payload)), + ) + + assert _probe_video_sample_end_us("ffprobe", tmp_path / "clip.mp4") == 2_001_000 + + +@pytest.mark.skipif( + shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="requires separately provisioned ffmpeg and ffprobe", +) +def test_exported_mp4_preserves_exact_pts_and_full_terminal_sample( + tmp_path: Path, +) -> None: + capture = _PresentationCapture(mode="replay") + capture.emitter.begin(profile="standard") + capture.emitter.emit_phase("observing", observation_png=_png((0, 0, 0))) + capture.emitter.emit_terminal("HALTED", observation_png=_png((255, 255, 255))) + capture.frames = [ + frame.model_copy(update={"observed_at_monotonic_ms": 1_000.0 + offset}) + for frame, offset in zip(capture.frames, (0, 250), strict=True) + ] + + _write_presentation_clip( + capture=capture, + pack_id="exact-timing-fixture", + clip_id="halted", + presentation_dir=tmp_path, + ) + + pts = json.loads((tmp_path / "halted.frame-pts-us.json").read_text()) + timeline = json.loads((tmp_path / "halted.control-overlay.v2.json").read_text()) + media = tmp_path / "halted.mp4" + assert pts["presentation_times_us"] == [0, 250_000, 2_000_000] + assert _probe_video_sample_end_us(str(shutil.which("ffprobe")), media) == 2_001_000 + assert timeline["duration_ms"] == 2_001 + + probe = subprocess.run( + [ + str(shutil.which("ffprobe")), + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=time_base:packet=pts,dts", + "-of", + "json", + str(media), + ], + check=True, + capture_output=True, + text=True, + ) + timing = json.loads(probe.stdout) + assert timing["streams"] == [{"time_base": "1/1000000"}] + assert all(packet["pts"] == packet["dts"] for packet in timing["packets"])