viewer: caret navigation, forward delete, numpad support, and accessibility deadline

src/phone_harness/config.py

@@ -10,7 +10,7 @@
 ENV_FILE = REPO_ROOT / ".env"
 
 WDA_PORT = 8100
-MJPEG_PORT = 9100
+MJPEG_PORT = 9100  # noqa: vulture
 
 WDA_URL = f"http://127.0.0.1:{WDA_PORT}"
 
@@ -40,20 +40,20 @@ def get(key: str, default: str | None = None) -> str | None:
 # Optional overrides
 # Read by device.detect_wda_bundle; else auto-detected from installed apps.
 WDA_BUNDLE_ID = get("WDA_BUNDLE_ID")  # noqa: vulture
-PHONE_PASSCODE = get("PHONE_PASSCODE")  # opt-in: lets helpers.unlock() type it
+PHONE_PASSCODE = get("PHONE_PASSCODE")  # noqa: vulture  (opt-in: lets helpers.unlock() type it)
 
 # Prompt-injection gate: always | flagged | off. The default for this machine;
 # the viewer's toggle writes .state/send_approval, which wins at send time.
 # Anything unrecognized falls back to "always" (approval.mode fails safe).
-SEND_APPROVAL = get("SEND_APPROVAL", "always")
+SEND_APPROVAL = get("SEND_APPROVAL", "always")  # noqa: vulture
 
 # How long a gated send waits for the human to click Approve in the viewer.
 # Running out is a denial, never a send: the safe direction is not sending.
-SEND_APPROVAL_TIMEOUT = float(get("SEND_APPROVAL_TIMEOUT", "120") or "120")
+SEND_APPROVAL_TIMEOUT = float(get("SEND_APPROVAL_TIMEOUT", "120") or "120")  # noqa: vulture
 
 # Default moved off 8765: Practical Systems' pipeline API owns that port on
 # this workstation, and both stacks must run at the same time.
-VIEWER_PORT = int(get("VIEWER_PORT", "8770") or "8770")
+VIEWER_PORT = int(get("VIEWER_PORT", "8770") or "8770")  # noqa: vulture
 
 # Post-gesture waits, applied by whichever client creates the shared session.
 # Measured on device: WDA's default animationCoolOffTimeout=2 made every swipe
@@ -72,4 +72,9 @@ def get(key: str, default: str | None = None) -> str | None:
 # poll (viewer will still call loadPhone() on load). Default 0 avoids the
 # viewer accidentally triggering a WDA wedge; operators can enable it in
 # .env with VIEWER_PHONE_POLL_SECONDS=10 for a 10s poll.
-VIEWER_PHONE_POLL_SECONDS = float(get("VIEWER_PHONE_POLL_SECONDS", "0") or "0")
+VIEWER_PHONE_POLL_SECONDS = float(get("VIEWER_PHONE_POLL_SECONDS", "0") or "0")  # noqa: vulture
+
+# Accessibility snapshot timeout (seconds). Added upstream in WDA #1214
+# (appium/WebDriverAgent#1214) to avoid indefinite hangs on apps with busy main
+# event loops (e.g. TikTok video feeds). 0 disables the bound; default is 2.0s.
+WDA_ACCESSIBILITY_DEADLINE = float(get("WDA_ACCESSIBILITY_DEADLINE", "2.0") or "2.0")

src/phone_harness/viewer.html

@@ -1132,7 +1132,13 @@ <h2 class="sec">Phone</h2>
 let keyBusy = false;
 const ARROW_KEYS = {
   ArrowLeft: '\uE012', ArrowRight: '\uE014',
-  ArrowUp: '\uE013', ArrowDown: '\uE015'
+  ArrowUp: '\uE013', ArrowDown: '\uE015',
+  Delete: '\uE017',
+  Home: '\uE011', End: '\uE010',
+  PageUp: '\uE00E', PageDown: '\uE00F'
+};
+const NUMPAD_MAP = {
+  Decimal: '.', Divide: '/', Multiply: '*', Subtract: '-', Add: '+', Separator: ','
 };
 let arrowBuf = [];
 let arrowBusy = false;
@@ -1149,6 +1155,7 @@ <h2 class="sec">Phone</h2>
   }
 }
 async function sendArrowKey(key) {
+  if (keyBuf) flushKeys();
   arrowBuf.push(key);
   if (arrowBusy) return;
   arrowBusy = true;
@@ -1209,7 +1216,9 @@ <h2 class="sec">Phone</h2>
   let ch = null;
   if (ev.key === 'Enter') ch = '\n';
   else if (ev.key === 'Backspace') ch = '\b';
+  else if (NUMPAD_MAP[ev.key]) ch = NUMPAD_MAP[ev.key];
   else if (ev.key.length === 1) ch = ev.key;           // letters, digits, space, emoji-less symbols
+  else if (ev.code && /^Numpad\d$/.test(ev.code)) ch = ev.code.slice(6);
   if (ch === null) return;
   ev.preventDefault();
   keyBuf += ch;

src/phone_harness/wda_client.py

@@ -126,7 +126,10 @@ def _activity_summary(path: str, payload: dict | None) -> str | None:
         return f"open app: {payload.get('bundleId', '?')}"
     if path.endswith("/actions"):
         try:
-            steps = payload["actions"][0]["actions"]
+            action_item = payload["actions"][0]
+            if action_item.get("type") == "key":
+                return "key press"
+            steps = action_item["actions"]
             moves = [s for s in steps if s.get("type") == "pointerMove"]
             pauses = [s.get("duration", 0) for s in steps if s.get("type") == "pause"]
             if len(moves) >= 2:
@@ -392,15 +395,16 @@ def _create_session(self) -> str:
         _write_shared_session(sid)
         try:
             # Settings ride with the session, so only the creator applies them.
+            settings = {
+                "waitForIdleTimeout": config.WDA_IDLE_WAIT,
+                "animationCoolOffTimeout": config.WDA_ANIM_COOLOFF,
+            }
+            if config.WDA_ACCESSIBILITY_DEADLINE > 0:
+                settings["accessibilityDeadline"] = config.WDA_ACCESSIBILITY_DEADLINE
             self._request(
                 "POST",
                 f"/session/{sid}/appium/settings",
-                {
-                    "settings": {
-                        "waitForIdleTimeout": config.WDA_IDLE_WAIT,
-                        "animationCoolOffTimeout": config.WDA_ANIM_COOLOFF,
-                    }
-                },
+                {"settings": settings},
             )
         except WDAError:
             pass  # a session on default waits is slow, not broken
@@ -509,6 +513,25 @@ def _pointer_actions(self, steps: list[dict]) -> None:
             },
         )
 
+    def key_press(self, key: str) -> None:  # noqa: vulture  (called by viewer.py)
+        """Press one non-printing key through W3C keyboard actions."""
+        self._session_request(
+            "POST",
+            "/actions",
+            {
+                "actions": [
+                    {
+                        "type": "key",
+                        "id": "keyboard1",
+                        "actions": [
+                            {"type": "keyDown", "value": key},
+                            {"type": "keyUp", "value": key},
+                        ],
+                    }
+                ]
+            },
+        )
+
     def tap(self, x: float, y: float) -> None:
         self._pointer_actions(
             [

tests/test_viewer.py

@@ -47,6 +47,15 @@ def lock(self):  # noqa: vulture  (duck-typed stand-in for WDAClient)
 
     clipboard_content = ""
 
+    def tap(self, x, y):  # noqa: vulture  (duck-typed stand-in for WDAClient)
+        self.calls.append(("tap", x, y))
+
+    def type_text(self, text):  # noqa: vulture  (duck-typed stand-in for WDAClient)
+        self.calls.append(("type_text", text))
+
+    def key_press(self, key):  # noqa: vulture  (duck-typed stand-in for WDAClient)
+        self.calls.append(("key_press", key))
+
     def get_clipboard(self):  # noqa: vulture  (duck-typed stand-in for WDAClient)
         return self.clipboard_content
 
@@ -1306,6 +1315,31 @@ def test_gesture_buttons_give_focus_back_to_the_phone():
     assert "input,textarea,button,select,[tabindex]" in html
 
 
+def test_arrow_keys_send_wda_text_caret_controls():
+    # Printable text uses /wda/keys, but caret navigation must use a W3C key
+    # action. The text endpoint inserts private-use values as content on this
+    # WDA build.
+    html = (Path(viewer.__file__).parent / "viewer.html").read_text(encoding="utf-8")
+    start = html.index("window.addEventListener('keydown'")
+    body = html[start : html.index("const JSON_HDR", start)]
+    assert "const ARROW_KEYS" in html
+    assert "ArrowLeft: '\\uE012'" in html
+    assert "ArrowRight: '\\uE014'" in html
+    assert "ArrowUp: '\\uE013'" in html
+    assert "ArrowDown: '\\uE015'" in html
+    assert "Delete: '\\uE017'" in html
+    assert "NUMPAD_MAP" in html
+    assert "sendArrowKey(arrow)" in body
+    assert "'/api/key'" in html
+    assert "text-cursor" not in html
+
+
+def test_key_endpoint_executes_key_press(base_url):
+    r = requests.post(base_url + "/api/key", json={"key": "\ue017"}, timeout=5)
+    assert r.status_code == 200 and r.json() == {"ok": True}
+    assert ("key_press", "\ue017") in viewer.Handler.client.calls
+
+
 def test_enter_sends_and_only_a_bare_enter_does():
     # Enter in the message box sends; Shift/Ctrl+Enter break the line. Ctrl is
     # the half that cannot be left to the browser: Chromium inserts NOTHING for

tests/test_wda_client.py

@@ -467,9 +467,11 @@ def test_new_session_applies_standard_action_waits(wda):
     assert FakeWDA.last_settings == {
         "waitForIdleTimeout": config.WDA_IDLE_WAIT,
         "animationCoolOffTimeout": config.WDA_ANIM_COOLOFF,
+        "accessibilityDeadline": config.WDA_ACCESSIBILITY_DEADLINE,
     }
     assert config.WDA_IDLE_WAIT == 2.0
     assert config.WDA_ANIM_COOLOFF == 0.0
+    assert config.WDA_ACCESSIBILITY_DEADLINE == 2.0
 
 
 def test_adopting_client_does_not_retune(wda):
@@ -649,3 +651,22 @@ def test_activity_summary_clipboard():
     )
     assert summary == "set clipboard (8 b64 chars)"
     assert "Hello" not in summary
+
+
+def test_activity_summary_key_press():
+    summary = _activity_summary(
+        "/session/s/actions",
+        {
+            "actions": [
+                {
+                    "type": "key",
+                    "id": "keyboard1",
+                    "actions": [
+                        {"type": "keyDown", "value": "\ue017"},
+                        {"type": "keyUp", "value": "\ue017"},
+                    ],
+                }
+            ]
+        },
+    )
+    assert summary == "key press"