diff --git a/packages/flow/CHANGELOG.md b/packages/flow/CHANGELOG.md index fea6cab853..8854402d95 100644 --- a/packages/flow/CHANGELOG.md +++ b/packages/flow/CHANGELOG.md @@ -43,6 +43,8 @@ ### Fixes +- Typed input now needs the complete whitespace-squashed OCR value, including values shorter than four characters. Native and pixel-only runs halt when delivery cannot be confirmed instead of clicking and typing again. The Windows agent checks the foreground window before and after each character or clipboard paste; a change stops delivery and reports that input may be partial ([#1478](https://github.com/OpenAdaptAI/OpenAdapt/issues/1478)). + - A REST read names its own records path: `effects.kind: rest`, a reward recipe of kind `rest`, `RestRecordVerifier`, `RestCompensator` and the fault gate's in-memory copy require `records_path` (and `records_key`, `delete_path_template`) instead of defaulting to the demo app's `/api/db`; `run --effects-kind rest` takes `--effects-records-path`. Compile-time REST binding admission (`compile_recording(admit_api_bindings=True, admission_fixture=...)`) takes the app's read path, write path, server-assigned fields and idempotency key field from an `AdmissionFixture`, and proposes nothing when no write was captured and no write path is named. Effect mining and `scaffold-verifier` mine an idempotency key only from a field the caller names. Verifier identities and fault-gate check digests of existing configs are unchanged. - A FHIR read names its resource and field map: `effects.kind: fhir`, a reward recipe of kind `fhir` and `FhirEffectVerifier` require `resource_type` and `field_paths` instead of defaulting to one app's note-as-Observation layout (`DEFAULT_OBSERVATION_PATHS` stays importable for callers whose app stores notes that way). An app profile's `no_record_values` lists the record-field values its audit entries use for "no record" (OpenEMR's profile: `"0"`; profile version `openemr-frame-profile/2026-11-03.1`, so a config that pins the old profile digest is refused until it pins the new one); the frame and rollback no longer treat `0` as "no record" in every app. `generate ensure-contract` reads only the generic server-fact names from `facts.json`; the Medplum probe's translation moved to `benchmark/fhir_medplum/readpath/facts.py`. diff --git a/packages/flow/openadapt_flow/backends/win_agent/README.md b/packages/flow/openadapt_flow/backends/win_agent/README.md index 81bc208efd..d311445afd 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/README.md +++ b/packages/flow/openadapt_flow/backends/win_agent/README.md @@ -25,6 +25,13 @@ Only the Python **standard library** is imported at load; `mss`/Pillow (for the screenshot) and `pyautogui`/`uiautomation` (used by the input handlers) import lazily. No Flask required in the guest. +Typing binds to the foreground window handle and process. The agent checks +that identity before and after each ASCII character or Unicode clipboard +paste. If it changes, delivery stops with `input_focus_changed`; the client +reports uncertain delivery because input may already be partial. When the +foreground cannot be read, the agent refuses before typing. Update the guest +agent to get these checks. + ## Security (TLS in transit + loopback default + bearer token) `/screenshot` returns a PNG of the live patient chart and the input routes drive diff --git a/packages/flow/openadapt_flow/backends/win_agent/server.py b/packages/flow/openadapt_flow/backends/win_agent/server.py index 0c889efe7c..240518d94d 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -449,8 +449,29 @@ def _perform_input(payload: dict[str, Any]) -> dict[str, Any]: pyautogui.FAILSAFE = False if text: + foreground = _foreground_window_identity() + if foreground is None: + raise AgentRequestError( + 503, + "input_focus_unavailable", + "cannot bind keyboard input to a foreground window", + ) + + def require_foreground() -> None: + if _foreground_window_identity() != foreground: + raise AgentRequestError( + 409, + "input_focus_changed", + "foreground window changed during typing; delivery may be partial", + ) + if text.isascii(): - pyautogui.write(text, interval=float(interval)) + for index, char in enumerate(text): + require_foreground() + pyautogui.write(char, interval=0, _pause=False) + require_foreground() + if index + 1 < len(text): + time.sleep(float(interval)) else: encoded = base64.b64encode(text.encode("utf-8")).decode("ascii") ps = ( @@ -465,7 +486,9 @@ def _perform_input(payload: dict[str, Any]) -> dict[str, Any]: if completed.returncode != 0: raise RuntimeError("Set-Clipboard failed") time.sleep(0.2) + require_foreground() pyautogui.hotkey("ctrl", "v") + require_foreground() return _delivery_receipt("physical_type_text", native=False) if action == "press": @@ -2280,8 +2303,8 @@ def _running_image_digest(pid: int, path: str) -> Optional[str]: return _image_file_digest(path) -def _foreground_process_id() -> Optional[int]: - """The process that owns the foreground window, without reading its title.""" +def _foreground_window_identity() -> Optional[tuple[int, int]]: + """Foreground window and owning process, without reading a window title.""" if os.name != "nt": return None @@ -2311,7 +2334,13 @@ def _foreground_process_id() -> Optional[int]: return None except Exception: # noqa: BLE001 - unavailable means unverifiable return None - return int(pid.value) + return int(hwnd), int(pid.value) + + +def _foreground_process_id() -> Optional[int]: + """The process that owns the foreground window, without reading its title.""" + identity = _foreground_window_identity() + return identity[1] if identity is not None else None def _foreground_application_identity() -> Optional[str]: diff --git a/packages/flow/openadapt_flow/backends/windows_backend.py b/packages/flow/openadapt_flow/backends/windows_backend.py index 846ac2cf34..9b69963ed2 100644 --- a/packages/flow/openadapt_flow/backends/windows_backend.py +++ b/packages/flow/openadapt_flow/backends/windows_backend.py @@ -498,6 +498,12 @@ def _post_typed_action(self, path: str, payload: dict) -> dict: "guarded Windows input refused " f"({code}): {message or 'identity binding changed'}" ) + if path in {"/input", "/input/guarded"} and code == "input_focus_changed": + raise ActionDeliveryUncertain( + operation=operation, + native=False, + cause_type="ForegroundChanged", + ) if response.status_code >= 500: raise ActionDeliveryUncertain( operation=operation, diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 3c0edc1235..f1c3180f4b 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -48,6 +48,7 @@ import threading import time import uuid +from collections import Counter from copy import deepcopy from dataclasses import dataclass from datetime import date, datetime, timezone @@ -85,6 +86,7 @@ RemoteFrameContractBackend, RichPointerActionBackend, SelectOptionBackend, + StructuralBackend, StructuralElementNotActionable, StructuralResolutionRefused, SubmitBindingBackend, @@ -395,13 +397,12 @@ def _frame_viewport(frame_png: bytes) -> tuple[int, int]: # modal at 0.24, leaving a deliberate margin on both sides. PC_STRUCTURAL_TEMPLATE_THRESHOLD = 0.8 -# Typed-input verification: size of the "field region" diffed/OCRed around -# the focusing click point after a TYPE action. Generous on purpose — the -# typed text renders at the field's own left edge / first line, not at the -# click point — and clamped to the viewport. When no focusing click is -# known (keyboard-only focus moves, e.g. Tab between fields), the whole -# frame is used instead. -FIELD_REGION_SIZE = (640, 240) +# Typed-input verification: height of the visual readback band around the +# focusing click after TYPE. Without structural field bounds, read the full +# viewport width: a centered crop can hide the value's beginning or suffix. +# With no focusing click (e.g. Tab between fields), use the whole frame. +FIELD_REGION_HEIGHT = 240 +MASKED_FIELD_REGION_WIDTH = 640 FIELD_REGION_PAD = 16 # Typed-input verification, secret-only diff acceptance: when a SECRET typed @@ -17476,6 +17477,8 @@ def _field_region( self, field_point: Optional[Point], structural_region: Optional[Region] = None, + *, + masked: bool = False, ) -> Optional[Region]: """Region to observe for typed input, or None for the whole frame.""" vw, vh = self.backend.viewport @@ -17489,8 +17492,8 @@ def _field_region( return (x, y, right - x, bottom - y) if field_point is None: return None - w = min(FIELD_REGION_SIZE[0], vw) - h = min(FIELD_REGION_SIZE[1], vh) + w = min(MASKED_FIELD_REGION_WIDTH, vw) if masked else vw + h = min(FIELD_REGION_HEIGHT, vh) x = min(max(0, field_point[0] - w // 2), max(0, vw - w)) y = min(max(0, field_point[1] - h // 2), max(0, vh - h)) return (x, y, w, h) @@ -17557,14 +17560,14 @@ def _typed_input_landed( """Did the just-typed ``text`` visibly land? Exact structural readback is authoritative when the backend exposes - it. Otherwise, for an OCR-able value (>= - ``identity.MIN_PARAM_CHARS`` squashed chars), a contiguous squashed - run of the value (retried at 2x resolution) must be readable in the - field region. Pixel-only acceptance is restricted to steps declared - ``secret``: masked password dots may render no readable text. A dialog + it. Otherwise the complete lowercased, whitespace-squashed value must + match the field's new OCR text (retried at 2x resolution), including + short values. + Whole unchanged context lines may be excluded. Pixel-only acceptance + is restricted to steps declared ``secret``: masked password dots may + render no readable text. A dialog painting over the region or an unreadable ordinary field must never - count as "input landed". Values too short for OCR to arbitrate retain - the legacy diff fallback. + count as "input landed". Returns: ``(landed, changed)`` — the verdict, and whether the field @@ -17572,7 +17575,9 @@ def _typed_input_landed( retyping is only safe when nothing changed). """ after_png = self._settle_seeded() - region = self._field_region(field_point, field_region) + # Masked input has no readable value: keep its pixel-only change + # evidence local, so an unrelated change elsewhere cannot confirm it. + region = self._field_region(field_point, field_region, masked=allow_masked) if baseline_field_value is not None: expected = baseline_field_value + text after_at_point = ( @@ -17627,21 +17632,45 @@ def _typed_input_landed( == identity_mod.squash(text).casefold() ): return True, True + if not changed: + return False, False needle = identity_mod.squash(text) - if len(needle) < identity_mod.MIN_PARAM_CHARS: - return changed, changed # too short for OCR to arbitrate - need = identity_mod.required_run(len(needle)) - after_hay = self._ocr_squashed(after_png, region) - if identity_mod.longest_run(needle, after_hay) >= need: + + def exact_ocr_value( + before: bytes, after: bytes, crop: Optional[Region] + ) -> bool: + observed = [ + identity_mod.squash(line.text) + for line in self.vision.ocr(after, region=crop) + if identity_mod.squash(line.text) + ] + if needle and "".join(observed) == needle: + return True + if allow_masked or not needle: + return False + # The readback window may include unchanged labels beside the + # field. Remove whole unchanged OCR lines, with multiplicity; + # every remaining character must belong to the complete value. + previous = Counter( + identity_mod.squash(line.text) + for line in self.vision.ocr(before, region=crop) + ) + added = [] + for value in observed: + if previous[value]: + previous[value] -= 1 + else: + added.append(value) + return "".join(added) == needle + + if exact_ocr_value(baseline_png, after_png, region): return True, changed if region is not None: upscaled = identity_mod.upscale_crop(after_png, region) - if upscaled is not None: - up_hay = self._ocr_squashed(upscaled, None) - if identity_mod.longest_run(needle, up_hay) >= need: + before_upscaled = identity_mod.upscale_crop(baseline_png, region) + if upscaled is not None and before_upscaled is not None: + if exact_ocr_value(before_upscaled, upscaled, None): return True, changed - if not changed: - return False, False # Pixels changed but the value is unreadable: masked rendering is # the only acceptable explanation, and masked rendering adds no # confidently readable alphanumeric text (dot glyphs OCR as @@ -17677,10 +17706,10 @@ def _verify_typed_input( ) -> Optional[str]: """Verify a TYPE action landed; one guarded refocus-and-retype retry. - The retry only fires when exact readback or pixels prove the first - attempt changed NOTHING in the field (keystrokes fell on a - non-rendering target): re-click - the field (when its point is known), select-all so a false-negative + A retry needs a browser surface, structural browser observations and + exact field readback proving no change. An opaque or native surface + cannot prove where the first input went and must halt. For a browser, + re-click the field (when its point is known), select-all so a false-negative first attempt is replaced rather than duplicated, retype. When the region DID change but the value cannot be read (a dialog over the field, an unexpected re-render), retyping is not safe — select-all @@ -17710,12 +17739,32 @@ def _verify_typed_input( "or instead of the input) — retyping is unsafe in this " "state; run aborted" ) - if not allow_retry: + can_retry = ( + allow_retry + and workflow.surface in (None, "web") + and isinstance(self.backend, StructuralBackend) + and bool(self.backend.url) + and baseline_field_value is not None + ) + if not can_retry: result.input_verified = False + # This is a halt, not a new delivery edge: recheck authority + # without consuming a durable fence or an approval permit. + refusal = self._fresh_actuation_authorization_refusal( + workflow, params, step + ) + if refusal is not None: + result.safety_halt = True + result.failure_category = ( + "governed_refusal" + if self.governed_authorization is not None + else "safety_halt" + ) + return refusal return ( f"Typed input could not be verified for step '{step.id}' " - f"({step.intent}): the identity-bound browser field showed no " - "confirmed change; refusing an unleased refocus/retype retry" + f"({step.intent}): the target field showed no confirmed change; " + "refusing an unleased refocus/retype retry after possibly stray input" ) result.input_retried = True retry_resolution = resolution diff --git a/packages/flow/tests/test_page_settle.py b/packages/flow/tests/test_page_settle.py index 3d4ac8879c..cd3c777c30 100644 --- a/packages/flow/tests/test_page_settle.py +++ b/packages/flow/tests/test_page_settle.py @@ -368,7 +368,7 @@ def _typed(label, text="Massachusetts"): r.backend = _SelectBackend(label) r.vision = _PixelVision() r._settle_seeded = lambda: b"after" - r._field_region = lambda point, region: (0, 0, 10, 10) + r._field_region = lambda point, region, *, masked=False: (0, 0, 10, 10) return r._typed_input_landed(text, (5, 5), b"before"), r.vision.ocr_calls diff --git a/packages/flow/tests/test_pixel_scale_replay.py b/packages/flow/tests/test_pixel_scale_replay.py index c55686a9d9..e701c0bbdc 100644 --- a/packages/flow/tests/test_pixel_scale_replay.py +++ b/packages/flow/tests/test_pixel_scale_replay.py @@ -90,6 +90,29 @@ def test_canvas_kiosk_at_150_percent_cleartype_off_books_the_right_record( def test_wrong_record_at_a_new_scale_still_halts_before_the_write( + tmp_path: Path, bundles: dict[str, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + kiosk = ScaledKiosk(SPECS["canvas_notes"](), RenderProfile(1.25, cleartype=True)) + verify_input = Replayer._verify_typed_input + + def switch_after_verified_input(self, *args, **kwargs): + error = verify_input(self, *args, **kwargs) + assert error is None, "the real input check must pass before the identity fault" + kiosk.select_patient(1) + return error + + # Exercise the next action's identity check independently of the TYPE + # check. A switch during readback also changes the field's context and + # must already halt there (covered separately below). + monkeypatch.setattr(Replayer, "_verify_typed_input", switch_after_verified_input) + report = _replay(tmp_path, bundles["canvas_notes"], kiosk) + assert not report.success + assert kiosk.state.saved == [] + failure = _first_failure(report) + assert failure.identity is not None and failure.identity.status == "mismatch" + + +def test_record_switch_during_typed_readback_halts_without_retry( tmp_path: Path, bundles: dict[str, Path] ) -> None: kiosk = ScaledKiosk(SPECS["canvas_notes"](), RenderProfile(1.25, cleartype=True)) @@ -101,7 +124,10 @@ def switch_record(k: ScaledKiosk, text: str) -> None: assert not report.success assert kiosk.state.saved == [] failure = _first_failure(report) - assert failure.identity is not None and failure.identity.status == "mismatch" + assert failure.input_verified is False + assert failure.input_retried is False + assert failure.safety_halt is True + assert kiosk.state.values["note"] == sweep.REPLAY["canvas_notes"]["note"] def test_unconfirmable_identity_at_a_new_scale_halts( diff --git a/packages/flow/tests/test_replayer.py b/packages/flow/tests/test_replayer.py index a0ca49a7fb..3dbe487a98 100644 --- a/packages/flow/tests/test_replayer.py +++ b/packages/flow/tests/test_replayer.py @@ -1552,7 +1552,7 @@ def test_typed_recovery_refuses_inputs_changed_during_verification(bundle, run_d result = report.results[0] assert report.success is False assert backend.actions == [("type", "hello")] - assert result.input_retried is True + assert result.input_retried is False assert result.input_verified is False assert "authorization no longer matches the current runtime inputs" in ( result.error or "" @@ -3813,6 +3813,7 @@ def test_visual_click_crop_does_not_narrow_following_type_verification(bundle, r Match(point=(210, 150), region=(20, 20, 12, 8), confidence=0.95) ] vision.pixels_changed_results = [True] + vision.ocr_lines = [OcrLine("MA")] backend = FakeBackend(text_value_supported=False) workflow = Workflow( name="wf", @@ -3910,6 +3911,65 @@ def test_type_verification_prefers_exact_structural_field_region() -> None: ) +@pytest.mark.parametrize( + "rendered,verified", + [ + ([(20, "7")], True), + ([(20, "X"), (900, "7")], False), + ([(900, "7"), (1400, "X")], False), + ], +) +def test_visual_input_readback_cannot_hide_text_outside_a_centered_crop( + rendered, verified +): + before = make_png((1600, 800)) + after = make_png((1600, 800), color=(220, 220, 220)) + + class CroppedVision(FakeVision): + def ocr(self, png, *, region=None): + if png == before or region is None: + return [] + x, _y, width, _height = region + text = "".join(value for left, value in rendered if x <= left < x + width) + return [OcrLine(text)] if text else [] + + backend = FakeBackend(frame=after, viewport=(1600, 800), text_value_supported=False) + landed, changed = Replayer(backend, vision=CroppedVision())._typed_input_landed( + "7", (900, 500), before + ) + assert landed is verified + assert changed is True + assert backend.actions == [] + + +@pytest.mark.parametrize("changed_x,expected", [(50, False), (900, True)]) +def test_masked_input_needs_a_change_near_the_target(changed_x, expected): + from PIL import ImageDraw + + from openadapt_flow.vision import pixels_changed + + before = make_png((1600, 800)) + image = Image.open(io.BytesIO(before)) + ImageDraw.Draw(image).rectangle( + (changed_x - 40, 450, changed_x + 40, 550), fill="black" + ) + stream = io.BytesIO() + image.save(stream, format="PNG") + after = stream.getvalue() + + class PixelVision(FakeVision): + def pixels_changed(self, before, after, *, region=None, **kwargs): + return pixels_changed(before, after, region=region) + + backend = FakeBackend(frame=after, viewport=(1600, 800), text_value_supported=False) + landed, changed = Replayer(backend, vision=PixelVision())._typed_input_landed( + "secret", (900, 500), before, allow_masked=True + ) + assert landed is expected + assert changed is expected + assert backend.actions == [] + + def test_type_verification_refocuses_and_retypes_once(bundle, run_dir): """Focus theft: the first attempt lands nowhere visible; the retry re-clicks the field, selects all (replace, never append), retypes, and @@ -3918,7 +3978,7 @@ def test_type_verification_refocuses_and_retypes_once(bundle, run_dir): vision.template_results = [ Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) ] - backend = FakeBackend(type_accept_results=[False, True]) + backend = StructuralFakeBackend(type_accept_results=[False, True]) workflow = Workflow( name="wf", steps=[ @@ -3951,7 +4011,7 @@ def test_type_verification_failure_halts_run(bundle, run_dir): vision.template_results = [ Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) ] - backend = FakeBackend(type_accept_results=[False, False]) + backend = StructuralFakeBackend(type_accept_results=[False, False]) workflow = Workflow( name="wf", steps=[ @@ -4011,6 +4071,180 @@ def test_type_verification_ocr_reads_the_value(bundle, run_dir): assert report.results[1].input_retried is False +@pytest.mark.parametrize( + "value,observed", + [ + ("1234567890", "12345678"), + ("A", "Wrong"), + ("12", "17"), + ("123", "Unrelated"), + ("alpha", "alphaX"), + ("alpha", "Xalpha"), + ("hello world", "hello worl"), + ], +) +def test_typed_input_requires_the_exact_ocr_value(bundle, run_dir, value, observed): + vision = _type_vision() + vision.ocr_lines = [OcrLine(observed)] + backend = FakeBackend(text_value_supported=False) + report = Replayer(backend, vision=vision).run( + _type_workflow(), + params={"note": value}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is False + result = report.results[1] + assert result.input_verified is False + assert result.input_retried is False + assert backend.actions.count(("type", value)) == 1 + assert ("press", "ControlOrMeta+a") not in backend.actions + + +@pytest.mark.parametrize("value,observed", [("7", "7"), ("Hello world", "hello WORLD")]) +def test_exact_ocr_input_including_short_values_is_verified( + bundle, run_dir, value, observed +): + vision = _type_vision() + vision.ocr_lines = [OcrLine(observed)] + backend = FakeBackend(text_value_supported=False) + report = Replayer(backend, vision=vision).run( + _type_workflow(), + params={"note": value}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is True + assert report.results[1].input_verified is True + assert report.results[1].input_retried is False + + +@pytest.mark.parametrize("exact_readback", [False, True]) +def test_opaque_typing_never_retries_an_unconfirmed_delivery( + bundle, run_dir, exact_readback +): + vision = _type_vision() + vision.pixels_changed_results = [False, False] + backend = FakeBackend( + text_value_supported=exact_readback, type_accept_results=[False, True] + ) + + class CountingDeliveryGuard: + before_calls = 0 + acknowledgments = 0 + + def before_delivery(self): + self.before_calls += 1 + + def acknowledge_delivery(self): + self.acknowledgments += 1 + + guard = CountingDeliveryGuard() + replayer = Replayer(backend, vision=vision) + replayer._durable_continuation_guard = guard + report = replayer.run( + _type_workflow(), + params={"note": "hello world"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is False + result = report.results[1] + assert result.input_verified is False + assert result.input_retried is False + assert backend.actions == [("click", 110, 105, False), ("type", "hello world")] + + assert guard.before_calls == guard.acknowledgments == len(backend.actions) == 2 + assert replayer._active_delivery_acknowledgers == () + + +def test_unavailable_browser_context_does_not_authorize_a_typing_retry(bundle, run_dir): + vision = _type_vision() + backend = StructuralFakeBackend(url=None, type_accept_results=[False, True]) + report = Replayer(backend, vision=vision).run( + _type_workflow(), + params={"note": "hello world"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is False + assert report.results[1].input_retried is False + assert backend.actions == [("click", 110, 105, False), ("type", "hello world")] + + +def test_an_opaque_surface_cannot_opt_into_browser_typing_recovery(bundle, run_dir): + from openadapt_flow.ir import StepResult + + backend = StructuralFakeBackend() + workflow = _type_workflow() + workflow.surface = "rdp" + result = StepResult(step_id="t1", intent="type note", ok=False) + error = Replayer(backend, vision=FakeVision())._verify_typed_input( + workflow.steps[1], + "hello world", + (110, 105), + backend.screenshot(), + result, + workflow=workflow, + params={"note": "hello world"}, + bundle_dir=bundle, + run_dir=run_dir, + resolution=None, + baseline_field_value="", + allow_retry=True, + ) + assert error is not None + assert result.input_verified is False + assert result.input_retried is False + assert backend.actions == [] + + +@pytest.mark.parametrize( + "observed,verified", + [ + (["Open field", "hello world"], True), + (["Open field", "hello", "world"], True), + (["Open field", "hello worl"], False), + (["Open field", "hello world", "extra"], False), + (["Open field", "extra hello world"], False), + ], +) +def test_ocr_context_never_excuses_a_partial_or_extra_value( + bundle, run_dir, observed, verified +): + vision = _type_vision() + after = [OcrLine(value) for value in observed] + before = [OcrLine("Open field")] + vision.ocr_results = [after, before, after, before] + backend = FakeBackend(text_value_supported=False) + report = Replayer(backend, vision=vision).run( + _type_workflow(), + params={"note": "hello world"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is verified + assert report.results[1].input_verified is verified + assert report.results[1].input_retried is False + assert backend.actions.count(("type", "hello world")) == 1 + + +def test_unchanged_pixels_cannot_verify_an_existing_ocr_value(bundle, run_dir): + vision = _type_vision() + vision.ocr_lines = [OcrLine("hello world")] + vision.pixels_changed_results = [False] + backend = FakeBackend(text_value_supported=False, type_accept_results=[False]) + report = Replayer(backend, vision=vision).run( + _type_workflow(), + params={"note": "hello world"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is False + assert report.results[1].input_verified is False + assert report.results[1].input_retried is False + + def test_type_dialog_over_field_halts_without_retyping(bundle, run_dir): """ADDED 2026-07-09 (adversarial review, P2a/P2b): a dialog rendering over the field region changes pixels — under the old diff-alone rule @@ -4138,8 +4372,8 @@ def test_type_masked_dots_reading_as_noise_still_accepts(bundle, run_dir, monkey def test_type_without_known_field_diffs_full_frame_and_cannot_refocus(bundle, run_dir): """A TYPE step not preceded by a click (keyboard-only focus moves) has - no field point: verification diffs the whole frame, and the retry - retypes without a refocus click.""" + no field point or structural browser readback: verification cannot + prove where input went, so it halts without a second delivery.""" vision = FakeVision() vision.pixels_changed_results = [False, False] backend = FakeBackend( @@ -4156,8 +4390,8 @@ def test_type_without_known_field_diffs_full_frame_and_cannot_refocus(bundle, ru ) assert report.success is False assert vision.pixels_changed_calls[0] is None # full-frame diff - # Retry typed again but never clicked or selected-all. - assert backend.actions == [("type", "North"), ("type", "North")] + assert report.results[0].input_retried is False + assert backend.actions == [("type", "North")] # -- structural postconditions (URL/title change, new tab) -------------------- diff --git a/packages/flow/tests/test_sanitized_artifact.py b/packages/flow/tests/test_sanitized_artifact.py index e365a652e9..e4ab761b5e 100644 --- a/packages/flow/tests/test_sanitized_artifact.py +++ b/packages/flow/tests/test_sanitized_artifact.py @@ -129,6 +129,7 @@ def _recording(tmp_path: Path) -> Path: class _RuntimeBackend: def __init__(self): self.actions: list[tuple[str, str]] = [] + self.value = "" @property def viewport(self): @@ -139,6 +140,10 @@ def screenshot(self): def type_text(self, value): self.actions.append(("type", value)) + self.value += value + + def focused_text_value(self): + return self.value def click(self, x, y, *, double=False): raise AssertionError("type-only parameter test must not click") diff --git a/packages/flow/tests/test_win_agent_server.py b/packages/flow/tests/test_win_agent_server.py index 61e860e421..6143f37eb9 100644 --- a/packages/flow/tests/test_win_agent_server.py +++ b/packages/flow/tests/test_win_agent_server.py @@ -1849,3 +1849,131 @@ def test_uia_tree_lists_one_windows_elements_without_values(monkeypatch) -> None ): with pytest.raises(AgentRequestError): _perform_uia("tree", bad) + + +@pytest.fixture +def physical_keyboard(monkeypatch): + """A physical-input seam with no desktop, clipboard or real key events.""" + from openadapt_flow.backends.win_agent import server + + state = types.SimpleNamespace( + foreground=(101, 202), typed="", pastes=0, after_write=None, after_paste=None + ) + fake = types.ModuleType("pyautogui") + + def write(text, **_kwargs): + state.typed += text + if state.after_write is not None: + state.after_write() + + def hotkey(*keys): + assert keys == ("ctrl", "v") + state.pastes += 1 + if state.after_paste is not None: + state.after_paste() + + fake.write = write + fake.hotkey = hotkey + monkeypatch.setitem(sys.modules, "pyautogui", fake) + monkeypatch.setattr(server, "_foreground_window_identity", lambda: state.foreground) + monkeypatch.setattr(server.time, "sleep", lambda _delay: None) + return state + + +def test_physical_typing_preserves_the_complete_value_with_stable_focus( + physical_keyboard, +): + receipt = _perform_input( + {"action": "type_text", "text": "safe-text", "interval_s": 0} + ) + assert physical_keyboard.typed == "safe-text" + assert receipt["status"] == "delivered" + assert receipt["operation"] == "physical_type_text" + + +def test_physical_typing_refuses_an_unobservable_foreground(physical_keyboard): + physical_keyboard.foreground = None + with pytest.raises(AgentRequestError, match="foreground window"): + _perform_input({"action": "type_text", "text": "abc"}) + assert physical_keyboard.typed == "" + assert physical_keyboard.pastes == 0 + + +@pytest.mark.parametrize("changed", [(303, 202), (101, 404), None]) +def test_physical_typing_stops_after_focus_changes(physical_keyboard, changed): + physical_keyboard.after_write = lambda: setattr( + physical_keyboard, "foreground", changed + ) + with pytest.raises(AgentRequestError, match="delivery may be partial") as error: + _perform_input({"action": "type_text", "text": "abc"}) + assert error.value.code == "input_focus_changed" + assert physical_keyboard.typed == "a" + + +def test_physical_typing_refuses_focus_change_before_the_first_key( + physical_keyboard, monkeypatch +): + from openadapt_flow.backends.win_agent import server + + readings = iter([(101, 202), (303, 202)]) + monkeypatch.setattr(server, "_foreground_window_identity", lambda: next(readings)) + with pytest.raises(AgentRequestError, match="delivery may be partial"): + _perform_input({"action": "type_text", "text": "abc"}) + assert physical_keyboard.typed == "" + + +def test_unicode_paste_is_refused_if_focus_changes_during_clipboard_setup( + physical_keyboard, monkeypatch +): + from openadapt_flow.backends.win_agent import server + + def clipboard(*_args, **_kwargs): + physical_keyboard.foreground = (303, 202) + return types.SimpleNamespace(returncode=0) + + monkeypatch.setattr(server.subprocess, "run", clipboard) + with pytest.raises(AgentRequestError, match="delivery may be partial"): + _perform_input({"action": "type_text", "text": "caf\u00e9"}) + assert physical_keyboard.typed == "" + assert physical_keyboard.pastes == 0 + + +def test_unicode_paste_reports_uncertainty_if_focus_changes_after_delivery( + physical_keyboard, monkeypatch +): + from openadapt_flow.backends.win_agent import server + + monkeypatch.setattr( + server.subprocess, "run", lambda *_a, **_k: types.SimpleNamespace(returncode=0) + ) + physical_keyboard.after_paste = lambda: setattr( + physical_keyboard, "foreground", (303, 202) + ) + with pytest.raises(AgentRequestError, match="delivery may be partial"): + _perform_input({"action": "type_text", "text": "caf\u00e9"}) + assert physical_keyboard.pastes == 1 + + +def test_focus_change_over_http_is_uncertain_and_never_uses_legacy_delivery( + physical_keyboard, +): + from openadapt_flow.backend import ActionDeliveryUncertain + from openadapt_flow.backends.windows_backend import WindowsBackend + + physical_keyboard.after_write = lambda: setattr( + physical_keyboard, "foreground", (303, 202) + ) + legacy_calls = [] + agent = RunningAgent( + AgentConfig(host="127.0.0.1", port=0, allow_legacy_exec=True), + legacy_fn=_recording_legacy_fn(legacy_calls), + ) + try: + backend = WindowsBackend(agent.url, allow_legacy_exec=True) + with pytest.raises(ActionDeliveryUncertain) as error: + backend.type_text("abc") + assert error.value.operation == "physical_type_text" + assert physical_keyboard.typed == "a" + assert legacy_calls == [] + finally: + agent.close()