perf(test): scale test concurrency and stop re-onboarding every Maestro flow

The Dart suite spent 77% of its cost compiling one isolate per test file
while `flutter test` used half the cores, and every Maestro flow replayed
a full Jellyfin onboarding before its first real assertion.

- Add scripts/run_tests.sh, which runs `flutter test` with -j set to the
  cores the process may actually use instead of the ncpu/2 default.
  Measured on 8 cores: 190s -> 136s; -j 12 regresses to 165s, so it scales
  to the core count rather than hard-coding one. CI and CONTRIBUTING use it.
  A cgroup v2 quota, a cgroup v1 quota, and the cpuset/affinity nproc
  reports can each be the binding limit independently, so the detector
  takes the smallest; trusting whichever it found first would oversubscribe
  4x on a container holding an 8-CPU quota while pinned to 2. Covered by
  scripts/test_run_tests.py, which the ci_guard_checks.sh glob picks up.
- Add .maestro/subflows/ensure_onboarded.yaml: cold-start the app and only
  onboard when no session is stored. Flows that just need a signed-in Home
  use it; 02_onboarding_home, 08_logout, 09_download_offline_playback and
  the profile regressions keep clearing state. 59s -> 16s per flow.
- Guard onboarding's two optional taps behind visibility checks. A missed
  `optional: true` tap still runs the full element search, costing 3.0s
  and 7.8s per onboarding to find nothing.
- Disable device animation scales in run_maestro.py, restored by the
  existing cleanup path. CI's emulator got this from the runner flag;
  physical devices never did.
- Shorten the watch_together setup-timeout replacement from 500ms to the
  10ms the same file already proves sufficient, and shorten the retry
  backoff at the one site that missed it: 8.04s -> 1.59s of execution.
- Make the LAN discovery waits deadline-based and resend the beacon while
  polling. Loopback UDP drops datagrams under load, which timed out a
  wait that could never be satisfied; this was the suite's one flaky test.
- Fix 08_logout, which searched for "Logout" and "Are you sure you want to
  logout?" after both strings became "Log out". The flow had been failing
  and aborting the suite before 09 ever ran.

flutter test 190s -> 131s. Maestro's Android suite 621s -> 385s across the
eight flows the baseline reached, and now runs all nine green.

.github/workflows/ci.yml

@@ -130,7 +130,7 @@ jobs:
       - name: Run tests
         run: |
           if [ -d "test" ] && [ "$(find test -name '*_test.dart' | wc -l)" -gt 0 ]; then
-            flutter test
+            scripts/run_tests.sh
           else
             echo "No tests found, skipping test execution"
           fi

.maestro/flows/03_library_browse.yaml

@@ -4,7 +4,7 @@ tags:
   - e2e
   - library
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^Libraries.*"
 - tapOn: "Browse"
 - extendedWaitUntil:

.maestro/flows/04_search.yaml

@@ -4,7 +4,7 @@ tags:
   - e2e
   - search
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^Search.*"
 - extendedWaitUntil:
     visible: "Search movies, shows, music..."

.maestro/flows/05_playback.yaml

@@ -4,7 +4,7 @@ tags:
   - e2e
   - playback
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^Libraries.*"
 - tapOn: "Browse"
 - extendedWaitUntil:

.maestro/flows/06_downloads.yaml

@@ -4,7 +4,7 @@ tags:
   - e2e
   - downloads
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^Downloads.*"
 - extendedWaitUntil:
     visible: "No downloads"

.maestro/flows/07_profiles_settings.yaml

@@ -5,7 +5,7 @@ tags:
   - profiles
   - settings
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^M(?:.*Profiles.*)?$"
 - tapOn:
     text: "Profiles"

.maestro/flows/08_logout.yaml

@@ -7,16 +7,16 @@ tags:
 - runFlow: ../subflows/onboard_jellyfin.yaml
 - tapOn: "(?s)^M(?:.*Profiles.*)?$"
 - tapOn:
-    text: "Logout"
+    text: "Log out"
     below:
       text: "Settings"
 - extendedWaitUntil:
-    visible: "Are you sure you want to logout?"
+    visible: "Are you sure you want to log out?"
     timeout: 10000
 - tapOn:
-    text: "Logout"
+    text: "Log out"
     below:
-      text: "Are you sure you want to logout?"
+      text: "Are you sure you want to log out?"
 - extendedWaitUntil:
     visible: "Sign in with Plex"
     timeout: 20000

.maestro/real_flows/01_media_catalog.yaml

@@ -5,7 +5,7 @@ tags:
   - real-jellyfin
   - media
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - tapOn: "(?s)^Search.*"
 - extendedWaitUntil:
     visible: "Search movies, shows, music..."

.maestro/real_flows/02_music_browse.yaml

@@ -7,7 +7,7 @@ tags:
   - music
   - regression
 ---
-- runFlow: ../subflows/onboard_jellyfin.yaml
+- runFlow: ../subflows/ensure_onboarded.yaml
 - assertVisible: "(?s)^Latest Albums in Maestro Music.*"
 - assertVisible: "(?s)^Regression Album.*"
 - tapOn: "(?s)^Libraries.*"

.maestro/subflows/ensure_onboarded.yaml

@@ -0,0 +1,27 @@
+appId: com.edde746.plezy
+---
+# Reach a signed-in Home without paying for onboarding that already happened.
+#
+# `onboard_jellyfin.yaml` clears app state and retypes the server URL and
+# credentials, which measures ~59s per flow on a Pixel 7 and dominates every
+# flow that only needs "signed in, on Home". A plain `launchApp` keeps stored
+# data and cold-starts straight back to Home, which measures ~16s.
+#
+# The guard keeps each flow runnable on its own: when no session is stored --
+# a fresh install, or a run after the logout flow -- "Discover" is absent and
+# this falls through to the real onboarding. Flows that must prove onboarding
+# itself, or that need pristine state, keep calling onboard_jellyfin.yaml.
+- launchApp
+# Settle on a known screen first. Branching straight off `launchApp` races the
+# splash: "Discover" has not rendered yet, the guard reads it as absent, and
+# every flow pays for a full re-onboarding it did not need.
+- extendedWaitUntil:
+    visible: "(?s)^(?:Discover|Sign in with Plex|Connect to Jellyfin|Wait).*"
+    timeout: 30000
+- runFlow:
+    when:
+      notVisible: "Discover"
+    file: onboard_jellyfin.yaml
+- extendedWaitUntil:
+    visible: "Discover"
+    timeout: 30000

.maestro/subflows/onboard_jellyfin.yaml

@@ -10,9 +10,15 @@ appId: com.edde746.plezy
       - extendedWaitUntil:
           visible: "(?s)^(?:Connect to Jellyfin|Wait).*"
           timeout: 30000
-      - tapOn:
+      # A miss on `tapOn ... optional: true` still runs Maestro's full element
-          text: "Wait"
+      # search before giving up, which measured 3s here and 7.8s for "Sign in"
-          optional: true
+      # on every onboarding. These taps normally find nothing, so gate them on
+      # a single visibility check instead.
+      - runFlow:
+          when:
+            visible: "Wait"
+          commands:
+            - tapOn: "Wait"
       - extendedWaitUntil:
           visible: "Connect to Jellyfin"
           timeout: 30000
@@ -34,9 +40,11 @@ appId: com.edde746.plezy
 - retry:
     maxRetries: 1
     commands:
-      - tapOn:
+      - runFlow:
-          text: "Sign in"
+          when:
-          optional: true
+            visible: "Sign in"
+          commands:
+            - tapOn: "Sign in"
       - extendedWaitUntil:
           visible: "Discover"
           timeout: 30000

.maestro/subflows/open_codec_sample.yaml

@@ -1,6 +1,6 @@
 appId: com.edde746.plezy
 ---
-- runFlow: onboard_jellyfin.yaml
+- runFlow: ensure_onboarded.yaml
 - tapOn: "(?s)^Search.*"
 - extendedWaitUntil:
     visible: "Search movies, shows, music..."

CONTRIBUTING.md

@@ -13,7 +13,7 @@
 - Run `dart format .` to format Dart code (note: generated files like `*.g.dart` are excluded from CI checks)
 - Run `scripts/format_native.sh --fix` to format Kotlin, Swift, C++, C, Objective-C, and native headers
 - Run `flutter analyze` before submitting to check for issues
-- Run `flutter test` if tests are available
+- Run `scripts/run_tests.sh` to run the test suite (same as `flutter test`, but scaled to your core count)
 - Test your changes thoroughly
 
 ### Code Quality Checks
@@ -34,7 +34,10 @@ The project includes automated CI checks that run on all pull requests:
    - Run locally: `scripts/codegen.sh --check`
 
 4. **Tests**: Runs unit and widget tests (when available)
-   - Run locally: `flutter test`
+   - Run locally: `scripts/run_tests.sh`
+   - This is `flutter test` with `-j` set to the core count. The default is half your cores, which
+     leaves most of the machine idle because the suite is dominated by per-file compilation.
+     Arguments are forwarded, so `scripts/run_tests.sh test/widgets/some_test.dart` works.
 
 All these checks must pass before your changes can be merged.
 

scripts/run_maestro.py

@@ -21,6 +21,11 @@ import urllib.request
 ROOT_DIR = Path(__file__).resolve().parent.parent
 APP_ID = "com.edde746.plezy"
 FAULTS = ("music-failure", "offline", "recovery")
+ANIMATION_SCALES = (
+    "window_animation_scale",
+    "transition_animation_scale",
+    "animator_duration_scale",
+)
 
 
 class RunnerError(RuntimeError):
@@ -547,6 +552,7 @@ class MaestroRunner:
             ("global", "stay_on_while_plugged_in"),
             ("secure", "immersive_mode_confirmations"),
             ("global", "hide_error_dialogs"),
+            *((("global", key) for key in ANIMATION_SCALES)),
         ):
             value = self._adb_capture("shell", "settings", "get", namespace, key)
             if value is not None:
@@ -572,6 +578,14 @@ class MaestroRunner:
             check=False,
             quiet=True,
         )
+        # Maestro waits for the view hierarchy to settle after every tap, input,
+        # and key press. With animations at their default 1.0 scale each of
+        # those waits pays for a real transition, which dominates a flow: taps
+        # measured 3-5s apiece on a physical Pixel 7. CI's emulator gets this
+        # from the runner's disable-animations flag; nothing was setting it for
+        # a real device. cleanup() restores the captured values.
+        for key in ANIMATION_SCALES:
+            self._adb_run("shell", "settings", "put", "global", key, "0", check=False, quiet=True)
         self._adb_run("shell", "input", "keyevent", "KEYCODE_BACK", check=False, quiet=True)
         _run_checked((*self.adb_prefix, "shell", "svc", "power", "stayon", "true"))
         _run_checked((*self.adb_prefix, "shell", "input", "keyevent", "KEYCODE_WAKEUP"))

scripts/run_tests.sh

@@ -0,0 +1,117 @@
+#!/usr/bin/env bash
+set -uo pipefail
+
+# Run the Flutter test suite with a concurrency that matches the host.
+#
+# `flutter test` defaults to ceil(numCPUs / 2), which leaves half the machine
+# idle. That default is a poor fit here because roughly three quarters of this
+# suite's cost is per-file Dart kernel compilation rather than test execution
+# (436 test files, each its own isolate), and compilation scales with cores.
+#
+# Measured on an 8-core host, full suite:
+#   -j 4 (the default)  190s
+#   -j 6                168s
+#   -j 8                136s
+#   -j 12               165s
+#
+# One job per core wins; oversubscribing regresses. So scale to the core count
+# instead of hard-coding a number that would oversubscribe smaller CI runners.
+#
+# Any arguments are forwarded to `flutter test`, and an explicit -j/--concurrency
+# still overrides the computed value.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+# Overridable so scripts/test_run_tests.py can point the detector at fixtures.
+: "${PLEZY_CGROUP_ROOT:=/sys/fs/cgroup}"
+
+# Cores this process may actually use.
+#
+# Three limits can each be the binding one, and a container can hit any subset
+# of them: a generous CPU quota paired with a narrow cpuset is as common as the
+# reverse. Taking whichever is discovered first would oversubscribe whenever a
+# different one binds, so collect them all and use the smallest.
+#
+#   cgroup v2 quota   cpu.max ("<quota> <period>", or "max" when unlimited)
+#   cgroup v1 quota   cpu.cfs_quota_us / cpu.cfs_period_us (-1 when unlimited)
+#   affinity/cpuset   reported by nproc, which honours sched_getaffinity
+online_cpus() {
+  if command -v nproc >/dev/null 2>&1; then
+    nproc 2>/dev/null && return
+  fi
+  if command -v sysctl >/dev/null 2>&1; then
+    sysctl -n hw.ncpu 2>/dev/null && return
+  fi
+  getconf _NPROCESSORS_ONLN 2>/dev/null
+}
+
+# ceil(quota / period), skipped unless both are positive integers.
+quota_cpus() {
+  local quota="$1" period="$2"
+  case "$quota$period" in
+    '' | *[!0-9]*) return 1 ;;
+  esac
+  [ "$period" -gt 0 ] || return 1
+  [ "$quota" -gt 0 ] || return 1
+  echo $(((quota + period - 1) / period))
+}
+
+detect_cpus() {
+  local limits=() value quota period
+
+  value="$(online_cpus)"
+  case "$value" in
+    '' | *[!0-9]*) ;;
+    *) limits+=("$value") ;;
+  esac
+
+  if [ -r "$PLEZY_CGROUP_ROOT/cpu.max" ]; then
+    read -r quota period <"$PLEZY_CGROUP_ROOT/cpu.max" || true
+    if value="$(quota_cpus "${quota:-}" "${period:-}")"; then
+      limits+=("$value")
+    fi
+  fi
+
+  if [ -r "$PLEZY_CGROUP_ROOT/cpu/cpu.cfs_quota_us" ] &&
+    [ -r "$PLEZY_CGROUP_ROOT/cpu/cpu.cfs_period_us" ]; then
+    read -r quota <"$PLEZY_CGROUP_ROOT/cpu/cpu.cfs_quota_us" || true
+    read -r period <"$PLEZY_CGROUP_ROOT/cpu/cpu.cfs_period_us" || true
+    if value="$(quota_cpus "${quota:-}" "${period:-}")"; then
+      limits+=("$value")
+    fi
+  fi
+
+  # Nothing readable anywhere: prefer a conservative guess over the host count.
+  if [ "${#limits[@]}" -eq 0 ]; then
+    echo 4
+    return
+  fi
+
+  local smallest="${limits[0]}"
+  for value in "${limits[@]}"; do
+    [ "$value" -lt "$smallest" ] && smallest="$value"
+  done
+  [ "$smallest" -lt 1 ] && smallest=1
+  echo "$smallest"
+}
+
+# Sourced by the tests to exercise the detector; only a direct run continues.
+if [ "${BASH_SOURCE[0]}" != "$0" ]; then
+  return 0
+fi
+
+cd "$ROOT"
+
+for arg in "$@"; do
+  case "$arg" in
+    -j | --concurrency | -j=* | --concurrency=*)
+      exec flutter test "$@"
+      ;;
+  esac
+done
+
+CPUS="$(detect_cpus)"
+
+echo "==> flutter test -j $CPUS ${*:-}"
+exec flutter test -j "$CPUS" "$@"

scripts/test_run_tests.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+"""Regression tests for the CPU detection in scripts/run_tests.sh.
+
+The detector picks the concurrency the whole suite runs at, and getting it wrong
+is silent: too high just makes CI slower. Every limit below can be the binding
+one independently, so each is pinned here against fixtures rather than trusted
+to whichever file happens to exist on the runner.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+RUN_TESTS = ROOT_DIR / "scripts" / "run_tests.sh"
+# Resolved up front: one test empties PATH, which would otherwise hide bash too.
+BASH = shutil.which("bash") or "/bin/bash"
+
+
+class DetectCpusTests(unittest.TestCase):
+    def detect(
+        self,
+        *,
+        nproc: int | None = None,
+        cpu_max: str | None = None,
+        cfs_quota: str | None = None,
+        cfs_period: str | None = None,
+        tools: bool = True,
+    ) -> str:
+        """Source run_tests.sh against fixtures and return detect_cpus output."""
+        with tempfile.TemporaryDirectory() as raw:
+            tmp = Path(raw)
+            bin_dir = tmp / "bin"
+            bin_dir.mkdir()
+            cgroup = tmp / "cgroup"
+            (cgroup / "cpu").mkdir(parents=True)
+
+            if nproc is not None:
+                stub = bin_dir / "nproc"
+                stub.write_text(f"#!/bin/sh\necho {nproc}\n", encoding="utf-8")
+                stub.chmod(0o755)
+
+            if cpu_max is not None:
+                (cgroup / "cpu.max").write_text(f"{cpu_max}\n", encoding="utf-8")
+            if cfs_quota is not None:
+                (cgroup / "cpu" / "cpu.cfs_quota_us").write_text(f"{cfs_quota}\n", encoding="utf-8")
+            if cfs_period is not None:
+                (cgroup / "cpu" / "cpu.cfs_period_us").write_text(f"{cfs_period}\n", encoding="utf-8")
+
+            # An empty PATH hides nproc/sysctl/getconf so the no-signal fallback
+            # is reachable; otherwise keep the real PATH so the stub shadows it.
+            path = f"{bin_dir}:{os.environ.get('PATH', '')}" if tools else str(bin_dir)
+            environment = {
+                **os.environ,
+                "PATH": path,
+                "PLEZY_CGROUP_ROOT": str(cgroup),
+            }
+            result = subprocess.run(
+                [BASH, "-c", f'source "{RUN_TESTS}"; detect_cpus'],
+                capture_output=True,
+                text=True,
+                check=True,
+                env=environment,
+            )
+            return result.stdout.strip()
+
+    def test_uses_the_online_cpu_count_when_uncapped(self) -> None:
+        self.assertEqual(self.detect(nproc=8), "8")
+
+    def test_cgroup_v2_quota_caps_a_larger_online_count(self) -> None:
+        # 200000/100000 == 2 CPUs of quota on an 8-core host.
+        self.assertEqual(self.detect(nproc=8, cpu_max="200000 100000"), "2")
+
+    def test_affinity_caps_a_larger_cgroup_v2_quota(self) -> None:
+        # The regression this suite exists for: a container can carry a quota
+        # worth 8 CPUs while being pinned to 2. Reading the quota alone and
+        # returning it would oversubscribe by 4x.
+        self.assertEqual(self.detect(nproc=2, cpu_max="800000 100000"), "2")
+
+    def test_unlimited_cgroup_v2_quota_falls_through_to_affinity(self) -> None:
+        self.assertEqual(self.detect(nproc=6, cpu_max="max 100000"), "6")
+
+    def test_cgroup_v1_quota_caps_a_larger_online_count(self) -> None:
+        self.assertEqual(
+            self.detect(nproc=8, cfs_quota="200000", cfs_period="100000"),
+            "2",
+        )
+
+    def test_unlimited_cgroup_v1_quota_falls_through_to_affinity(self) -> None:
+        self.assertEqual(
+            self.detect(nproc=6, cfs_quota="-1", cfs_period="100000"),
+            "6",
+        )
+
+    def test_smallest_limit_wins_when_both_cgroup_versions_are_present(self) -> None:
+        self.assertEqual(
+            self.detect(
+                nproc=16,
+                cpu_max="800000 100000",
+                cfs_quota="300000",
+                cfs_period="100000",
+            ),
+            "3",
+        )
+
+    def test_partial_quota_rounds_up(self) -> None:
+        # 2.5 CPUs of quota should not truncate to 2 and waste half a core.
+        self.assertEqual(self.detect(nproc=8, cpu_max="250000 100000"), "3")
+
+    def test_sub_single_core_quota_floors_at_one(self) -> None:
+        self.assertEqual(self.detect(nproc=8, cpu_max="50000 100000"), "1")
+
+    def test_malformed_quota_is_ignored_rather_than_trusted(self) -> None:
+        self.assertEqual(self.detect(nproc=8, cpu_max="garbage"), "8")
+        self.assertEqual(self.detect(nproc=8, cfs_quota="", cfs_period="100000"), "8")
+        self.assertEqual(self.detect(nproc=8, cpu_max="200000 0"), "8")
+
+    def test_falls_back_conservatively_with_no_signal_at_all(self) -> None:
+        self.assertEqual(self.detect(tools=False), "4")
+
+
+class RunTestsInvocationTests(unittest.TestCase):
+    def run_with_fake_flutter(self, arguments: list[str]) -> str:
+        """Run the script with `flutter` stubbed so it echoes its own argv."""
+        with tempfile.TemporaryDirectory() as raw:
+            tmp = Path(raw)
+            bin_dir = tmp / "bin"
+            bin_dir.mkdir()
+            cgroup = tmp / "cgroup"
+            (cgroup / "cpu").mkdir(parents=True)
+
+            flutter = bin_dir / "flutter"
+            flutter.write_text('#!/bin/sh\necho "FLUTTER $*"\n', encoding="utf-8")
+            flutter.chmod(0o755)
+            nproc = bin_dir / "nproc"
+            nproc.write_text("#!/bin/sh\necho 8\n", encoding="utf-8")
+            nproc.chmod(0o755)
+
+            result = subprocess.run(
+                [str(RUN_TESTS), *arguments],
+                capture_output=True,
+                text=True,
+                check=True,
+                env={
+                    **os.environ,
+                    "PATH": f"{bin_dir}:{os.environ.get('PATH', '')}",
+                    "PLEZY_CGROUP_ROOT": str(cgroup),
+                },
+            )
+            return result.stdout
+
+    def test_injects_the_detected_concurrency(self) -> None:
+        self.assertIn("FLUTTER test -j 8", self.run_with_fake_flutter([]))
+
+    def test_forwards_extra_arguments(self) -> None:
+        output = self.run_with_fake_flutter(["test/widgets/example_test.dart"])
+        self.assertIn("FLUTTER test -j 8 test/widgets/example_test.dart", output)
+
+    def test_explicit_concurrency_is_not_overridden(self) -> None:
+        for flag in (["-j", "2"], ["--concurrency=2"]):
+            with self.subTest(flag=flag):
+                output = self.run_with_fake_flutter(flag)
+                self.assertIn(f"FLUTTER test {' '.join(flag)}", output)
+                self.assertNotIn("-j 8", output)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/services/companion_remote_lan_discovery_service_test.dart

@@ -14,11 +14,17 @@ void main() {
       final listener = await _DiscoveryListener.start([context]);
 
       try {
-        await listener.sendBeacon(context: context, ips: const ['192.0.2.10']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.length == 1);
+          condition: () => listener.emissions.length == 1,
+          context: context,
+          ips: const ['192.0.2.10'],
+        );
 
-        await listener.sendBeacon(context: context, ips: const ['192.0.2.30', '10.0.0.30']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.length == 2);
+          condition: () => listener.emissions.length == 2,
+          context: context,
+          ips: const ['192.0.2.30', '10.0.0.30'],
+        );
 
         final hosts = listener.emissions.last;
         expect(hosts, hasLength(1));
@@ -38,12 +44,21 @@ void main() {
       final listener = await _DiscoveryListener.start([context]);
 
       try {
-        await listener.sendBeacon(context: context, platform: 'macOS', ips: const ['192.0.2.40', '10.0.0.40']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.length == 1);
+          condition: () => listener.emissions.length == 1,
+          context: context,
+          platform: 'macOS',
+          ips: const ['192.0.2.40', '10.0.0.40'],
+        );
 
+        // Reordered IPs only: must be absorbed without publishing an emission.
         await listener.sendBeacon(context: context, platform: 'macOS', ips: const ['10.0.0.40', '192.0.2.40']);
-        await listener.sendBeacon(context: context, platform: 'Android', ips: const ['192.0.2.40', '10.0.0.40']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.any((hosts) => hosts.single.platform == 'Android'));
+          condition: () => listener.emissions.any((hosts) => hosts.single.platform == 'Android'),
+          context: context,
+          platform: 'Android',
+          ips: const ['192.0.2.40', '10.0.0.40'],
+        );
 
         expect(listener.emissions, hasLength(2));
         final hosts = listener.emissions.last;
@@ -63,12 +78,21 @@ void main() {
       final listener = await _DiscoveryListener.start([firstContext, secondContext]);
 
       try {
-        await listener.sendBeacon(context: firstContext, name: 'Living Room', ips: const ['192.0.2.50']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.length == 1);
+          condition: () => listener.emissions.length == 1,
+          context: firstContext,
+          name: 'Living Room',
+          ips: const ['192.0.2.50'],
+        );
 
+        // Context churn only: must be absorbed without publishing an emission.
         await listener.sendBeacon(context: secondContext, name: 'Living Room', ips: const ['192.0.2.50']);
-        await listener.sendBeacon(context: secondContext, name: 'Living Room TV', ips: const ['192.0.2.50']);
+        await listener.sendBeaconUntil(
-        await _waitFor(() => listener.emissions.any((hosts) => hosts.single.name == 'Living Room TV'));
+          condition: () => listener.emissions.any((hosts) => hosts.single.name == 'Living Room TV'),
+          context: secondContext,
+          name: 'Living Room TV',
+          ips: const ['192.0.2.50'],
+        );
 
         expect(listener.emissions, hasLength(2));
         final hosts = listener.emissions.last;
@@ -96,6 +120,14 @@ RemoteAuthContext _authContext({required String id, required List<int> discovery
   );
 }
 
+// Loopback UDP needs a real poll: there is no seam to drive the socket's
+// receive path from fake time. Keep the interval short so a wait costs about
+// what the round trip costs, and bound every wait by a deadline rather than an
+// attempt count so the budget stays fixed if the interval changes.
+const _pollInterval = Duration(milliseconds: 2);
+const _resendInterval = Duration(milliseconds: 250);
+const _waitBudget = Duration(seconds: 5);
+
 class _DiscoveryListener {
   _DiscoveryListener._({
     required this.service,
@@ -129,13 +161,13 @@ class _DiscoveryListener {
     }
   }
 
-  Future<void> sendBeacon({
+  List<int> _encodeBeacon({
     required RemoteAuthContext context,
     required List<String> ips,
-    String name = 'Living Room',
+    required String name,
-    String platform = 'macOS',
+    required String platform,
-    int port = 52100,
+    required int port,
-  }) async {
+  }) {
     const version = 1;
     final auth = RemoteAuthService.instance;
     final homeHash = auth.computeDiscoveryTag(context.discoveryKey);
@@ -149,7 +181,7 @@ class _DiscoveryListener {
       port: port,
       ips: ips,
     );
-    final packet = utf8.encode(
+    return utf8.encode(
       jsonEncode({
         'app': 'plezy',
         'v': version,
@@ -162,14 +194,55 @@ class _DiscoveryListener {
         'hmac': hmac,
       }),
     );
+  }
 
-    for (var attempt = 0; attempt < 100; attempt++) {
+  Future<void> _transmit(List<int> packet) async {
+    final deadline = DateTime.now().add(_waitBudget);
+    while (DateTime.now().isBefore(deadline)) {
       if (sender.send(packet, InternetAddress.loopbackIPv4, service.discoveryPort) == packet.length) return;
-      await Future<void>.delayed(const Duration(milliseconds: 20));
+      await Future<void>.delayed(_pollInterval);
     }
     fail('Timed out sending LAN discovery beacon');
   }
 
+  Future<void> sendBeacon({
+    required RemoteAuthContext context,
+    required List<String> ips,
+    String name = 'Living Room',
+    String platform = 'macOS',
+    int port = 52100,
+  }) {
+    return _transmit(_encodeBeacon(context: context, ips: ips, name: name, platform: platform, port: port));
+  }
+
+  /// Sends a beacon repeatedly until [condition] holds.
+  ///
+  /// Loopback UDP drops datagrams when the host is loaded, and a lost beacon
+  /// leaves a plain [_waitFor] polling for an emission that can never arrive —
+  /// the observed flake in this suite. Re-sending is safe because the payload
+  /// is byte-identical and the service suppresses beacons that carry no
+  /// change, so only the first datagram to land can publish an emission.
+  Future<void> sendBeaconUntil({
+    required bool Function() condition,
+    required RemoteAuthContext context,
+    required List<String> ips,
+    String name = 'Living Room',
+    String platform = 'macOS',
+    int port = 52100,
+  }) async {
+    final packet = _encodeBeacon(context: context, ips: ips, name: name, platform: platform, port: port);
+    final deadline = DateTime.now().add(_waitBudget);
+    while (DateTime.now().isBefore(deadline)) {
+      await _transmit(packet);
+      final resendAt = DateTime.now().add(_resendInterval);
+      while (DateTime.now().isBefore(resendAt)) {
+        if (condition()) return;
+        await Future<void>.delayed(_pollInterval);
+      }
+    }
+    fail('Timed out waiting for LAN discovery behavior');
+  }
+
   Future<void> close() async {
     sender.close();
     await subscription.cancel();
@@ -178,9 +251,10 @@ class _DiscoveryListener {
 }
 
 Future<void> _waitFor(bool Function() condition) async {
-  for (var attempt = 0; attempt < 100; attempt++) {
+  final deadline = DateTime.now().add(_waitBudget);
+  while (DateTime.now().isBefore(deadline)) {
     if (condition()) return;
-    await Future<void>.delayed(const Duration(milliseconds: 20));
+    await Future<void>.delayed(_pollInterval);
   }
   fail('Timed out waiting for LAN discovery behavior');
 }

test/watch_together/watch_together_peer_service_test.dart

@@ -30,7 +30,7 @@ Future<T> _withShortenedTimer<T>({
 Future<T> _withSetupTimersShortened<T>(Future<T> Function() body) {
   return _withShortenedTimer(
     original: const Duration(seconds: 10),
-    replacement: const Duration(milliseconds: 500),
+    replacement: const Duration(milliseconds: 10),
     body: () => _withShortenedTimer(
       original: const Duration(milliseconds: 250),
       replacement: const Duration(milliseconds: 1),
@@ -531,11 +531,7 @@ void main() {
     final timeoutService = serviceFor(timeoutRelay);
 
     await expectLater(
-      _withShortenedTimer(
+      _withSetupTimersShortened(() => timeoutService.createSession(sessionId: 'slow1')),
-        original: const Duration(seconds: 10),
-        replacement: const Duration(milliseconds: 10),
-        body: () => timeoutService.createSession(sessionId: 'slow1'),
-      ),
       throwsA(
         isA<PeerError>()
             .having((error) => error.type, 'type', PeerErrorType.timeout)