diff --git a/AGENTS.md b/AGENTS.md
index 8204672aaec..caeafc6aa6e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -107,3 +107,55 @@ python butler.py format
 
 This will format the changed code in your current branch.
 It's possible to get into a state where linting and formatting contradict each other. In this case, STOP, the human will fix it.
+
+## Remote Swarming Development & Validation Workflow
+
+This section outlines the step-by-step cycle for developing, deploying, launching, and verifying changes on remote Swarming bots.
+
+### Step 1: Local Development & Formatting
+Make your code changes in your feature branch. Always run formatting and linting locally before committing:
+```bash
+pipenv run python butler.py format
+pipenv run python butler.py lint
+```
+
+### Step 2: Push & Deploy to Remote `dev`
+For changes to run on remote Swarming bots, they must be committed, merged, and pushed to the remote **`dev`** branch:
+1. Commit and push your feature branch:
+   ```bash
+   git add <modified_files>
+   git commit -m "Your description"
+   git push origin <your-feature-branch>
+   ```
+2. Merge into `dev` and push:
+   ```bash
+   git checkout dev
+   git pull origin dev
+   git merge <your-feature-branch>
+   git push origin dev
+   git checkout <your-feature-branch>
+   ```
+3. **⚠️ Crucial Rebuild Wait Time**: After pushing to `dev`, you **MUST wait 25 to 30 minutes** before triggering any Swarming tasks. This gives the remote Google Cloud Storage (GCS) builder enough time to pull your new commit, compile the binaries, and package them into the deployment ZIP bundle (`linux-3.zip`) fetched by the bots.
+
+### Step 3: Preprocess & Launch the Swarming Task
+Once the deployment package has finished rebuilding on GCS:
+1. Trigger the preprocess pipeline and launch a new Swarming task:
+   ```bash
+   python3 scratch/preprocess_and_launch.py
+   ```
+2. Note the generated **Swarming Task ID** (e.g. `791f445b26114a10`) and the task URL printed in the stdout.
+
+### Step 4: Live Monitoring & Log Retrieval
+1. **Live Monitor**: Open `scratch/monitor_swarming_task.py`, update `task_id` with your new Task ID, and run the script to stream the live console output:
+   ```bash
+   python3 scratch/monitor_swarming_task.py
+   ```
+2. **Download High-Resolution Logs**: Once the task terminates:
+   * Look at the live monitor output to identify the **assigned Bot Name** (e.g. `lin-192-g582`) and the final state (`COMPLETED` or `BOT_DIED`).
+   * Open `scratch/read_logs.py`, update the `bot_name` and adjust the time filter window (e.g. `timestamp >= "YYYY-MM-DDTHH:MM:SSZ"`), then run:
+     ```bash
+     pipenv run python scratch/read_logs.py
+     ```
+   * This downloads all high-resolution Stackdriver bot and logcat streams into `scratch/bot_logs.txt`.
+3. **Analyze**: Inspect `scratch/bot_logs.txt` using grep/editors to verify your changes (such as JNI prints or fuzzer loop outputs).
+
diff --git a/src/clusterfuzz/_internal/batch/service.py b/src/clusterfuzz/_internal/batch/service.py
index 0721d94e717..db933aa8761 100644
--- a/src/clusterfuzz/_internal/batch/service.py
+++ b/src/clusterfuzz/_internal/batch/service.py
@@ -256,7 +256,8 @@ def is_remote_task(command: str, job_name: str) -> bool:
   """
   try:
     _get_specs_from_config(
-        [remote_task_types.RemoteTask(command, job_name, None)])
+        [remote_task_types.RemoteTask(command, job_name, None)],
+        should_check_regions=False)
     return True
   except ValueError:
     return False
@@ -300,15 +301,15 @@ def _get_config_names(batch_tasks: List[remote_task_types.RemoteTask]):
   return config_map
 
 
-def _get_subconfig(batch_config, instance_spec):
+def _get_subconfig(batch_config, instance_spec, should_check_regions=True):
   all_subconfigs = batch_config.get('subconfigs', {})
   instance_subconfigs = instance_spec['subconfigs']
 
   queue_check_regions = batch_config.get('queue_check_regions')
-  if not queue_check_regions:
-    logs.info(
-        'Skipping batch load check because queue_check_regions is not configured.'
-    )
+  if not should_check_regions or not queue_check_regions:
+    logs.info('Skipping batch load check. '
+              f'should_check_regions: {should_check_regions}, '
+              f'queue_check_regions: {queue_check_regions}')
     weighted_subconfigs = [
         WeightedSubconfig(subconfig['name'], subconfig['weight'])
         for subconfig in instance_subconfigs
@@ -343,8 +344,8 @@ def _get_subconfig(batch_config, instance_spec):
   return all_subconfigs[chosen_name]
 
 
-def _get_specs_from_config(
-    batch_tasks: List[remote_task_types.RemoteTask]) -> Dict:
+def _get_specs_from_config(batch_tasks: List[remote_task_types.RemoteTask],
+                           should_check_regions: bool = True) -> Dict:
   """Gets the configured specifications for a batch workload."""
   if not batch_tasks:
     return {}
@@ -382,7 +383,8 @@ def _get_specs_from_config(
     # This saves us time and reduces fragementation, e.g. every linux fuzz task
     # run in this call will run in the same zone.
     if config_name not in subconfig_map:
-      subconfig = _get_subconfig(batch_config, instance_spec)
+      subconfig = _get_subconfig(batch_config, instance_spec,
+                                 should_check_regions)
       subconfig_map[config_name] = subconfig
 
     should_retry = instance_spec.get('retry', False)
diff --git a/src/clusterfuzz/_internal/bot/fuzzers/engine_common.py b/src/clusterfuzz/_internal/bot/fuzzers/engine_common.py
index b6881d86e5a..799e9a044bc 100644
--- a/src/clusterfuzz/_internal/bot/fuzzers/engine_common.py
+++ b/src/clusterfuzz/_internal/bot/fuzzers/engine_common.py
@@ -290,7 +290,9 @@ def find_fuzzer_path(build_directory, fuzzer_name):
   for root, _, files in shell.walk(build_directory):
     for filename in files:
       if (legacy_name_prefix + filename == fuzzer_name or
-          filename == fuzzer_filename):
+          filename == fuzzer_filename or
+          (filename.endswith('.apk') and fuzzer_name in filename) or
+          filename.endswith('.apk')):
         return os.path.join(root, filename)
 
   # This is an expected case when doing regression testing with old builds
diff --git a/src/clusterfuzz/_internal/bot/fuzzers/libfuzzer.py b/src/clusterfuzz/_internal/bot/fuzzers/libfuzzer.py
index 952b06cbe7d..84e3b2f4b18 100644
--- a/src/clusterfuzz/_internal/bot/fuzzers/libfuzzer.py
+++ b/src/clusterfuzz/_internal/bot/fuzzers/libfuzzer.py
@@ -20,6 +20,7 @@
 import random
 import re
 import shutil
+import subprocess
 
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.bot import testcase_manager
@@ -1122,6 +1123,451 @@ def cleanse_crash(self,
       return result
 
 
+class AndroidApkLibFuzzerRunner(new_process.UnicodeProcessRunner,
+                                LibFuzzerCommon):
+  """Android APK libFuzzer runner."""
+
+  def __init__(self, executable_path, build_directory, default_args=None):
+    super().__init__(
+        executable_path=android.adb.get_adb_path(), default_args=[])
+    self.apk_path = executable_path
+    self.package_name = android.app.get_package_name(self.apk_path)
+    if not self.package_name:
+      raise LibFuzzerError(f'Failed to get package name for {self.apk_path}')
+
+    android.app.install(self.apk_path)
+    self.instrumentation_runner = self._get_instrumentation_runner(
+        self.apk_path)
+    self.launchable_activity = self._get_launchable_activity(self.apk_path)
+
+    self.fuzzer_name = os.path.basename(self.apk_path).replace(
+        '-debug.apk', '').replace('.apk', '')
+    self.library_under_test = None
+    self.auxiliary_libraries = None
+
+    self._setup_dependencies(build_directory)
+
+  def _setup_dependencies(self, build_directory):
+    """Resolve, sort, and push C++ dependencies to the device."""
+    deps_filename = f"{self.fuzzer_name}__test_runner_script.runtime_deps"
+    deps_file_path = None
+    for root, _, files in os.walk(build_directory):
+      if deps_filename in files:
+        deps_file_path = os.path.join(root, deps_filename)
+        break
+
+    if not deps_file_path:
+      logs.info(f"No runtime_deps file found for {self.fuzzer_name}, "
+                "skipping dependency setup.")
+      return
+
+    logs.info(f"Found runtime_deps at: {deps_file_path}")
+
+    # Determine base output dir (parent of gen.runtime)
+    base_dir = deps_file_path
+    while True:
+      parent = os.path.dirname(base_dir)
+      if parent == base_dir:
+        break
+      if os.path.basename(base_dir) == 'gen.runtime':
+        base_dir = parent
+        break
+      base_dir = parent
+
+    # Read .so files from runtime_deps
+    libs = []
+    with open(deps_file_path, "r") as f:
+      for line in f:
+        line = line.strip()
+        if line.endswith(".so"):
+          abs_path = os.path.abspath(os.path.join(base_dir, line))
+          if os.path.exists(abs_path):
+            libs.append(abs_path)
+          else:
+            # Fallback: check if the library exists in the base_dir (stripped)
+            fallback_path = os.path.join(base_dir, os.path.basename(line))
+            if os.path.exists(fallback_path):
+              libs.append(fallback_path)
+            else:
+              logs.warning(
+                  f"Dependency file {abs_path} (and fallback {fallback_path}) "
+                  "does not exist.")
+
+    if not libs:
+      return
+
+    # Parse dependencies for each library
+    dependencies_dict = {}
+    for lib in libs:
+      lib_name = os.path.basename(lib)
+      dependencies_dict[lib_name] = self._parse_needed_libs(lib)
+
+    # Sort libraries topologically
+    sorted_libs = self._topological_sort(libs, dependencies_dict)
+
+    # Prepare device directory
+    android.adb.run_shell_command('rm -rf /sdcard/chromium_tests_root && '
+                                  'mkdir -p /sdcard/chromium_tests_root')
+
+    # Push sorted .so files to device
+    for lib in sorted_libs:
+      device_path = os.path.join('/sdcard/chromium_tests_root',
+                                 os.path.basename(lib))
+      android.adb.copy_local_file_to_remote(lib, device_path)
+
+    # Separate main library and auxiliary libraries
+    main_lib_name = f"lib{self.fuzzer_name}__library.so"
+    aux_libs = []
+    for lib in sorted_libs:
+      name = os.path.basename(lib)
+      if name == main_lib_name:
+        self.library_under_test = name
+      else:
+        aux_libs.append(name)
+
+    if aux_libs:
+      self.auxiliary_libraries = ','.join(aux_libs)
+
+    # Grant MANAGE_EXTERNAL_STORAGE permission
+    android.adb.run_shell_command(
+        f'appops set {self.package_name} MANAGE_EXTERNAL_STORAGE allow')
+    logs.info(f"Dependency setup complete. LibraryUnderTest: "
+              f"{self.library_under_test}, AuxiliaryLibraries: "
+              f"{self.auxiliary_libraries}")
+
+  def _parse_needed_libs(self, so_path):
+    """Parse NEEDED libraries using readelf."""
+    try:
+      res = subprocess.run(
+          ["readelf", "-d", so_path], capture_output=True, check=True)
+      output = res.stdout.decode('utf-8')
+      needed = []
+      for line in output.splitlines():
+        if "(NEEDED)" in line:
+          match = re.search(r"Shared library: \[(.*?)\]", line)
+          if match:
+            needed.append(match.group(1))
+      return needed
+    except Exception as e:
+      logs.error(f"Error reading ELF headers for {so_path}: {e}")
+      return []
+
+  def _topological_sort(self, libs, dependencies_dict):
+    """Perform topological sort on libs based on dependencies_dict."""
+    visited = {}
+    result = []
+
+    def visit(lib):
+      lib_name = os.path.basename(lib)
+      if visited.get(lib_name) == "visiting":
+        return
+      if visited.get(lib_name) == "visited":
+        return
+
+      visited[lib_name] = "visiting"
+      for dep in dependencies_dict.get(lib_name, []):
+        dep_path = None
+        for l in libs:
+          if os.path.basename(l) == dep:
+            dep_path = l
+            break
+        if dep_path:
+          visit(dep_path)
+
+      visited[lib_name] = "visited"
+      result.append(lib)
+
+    for lib in libs:
+      visit(lib)
+    return result
+
+  def _get_launchable_activity(self, apk_path):
+    """Return the launchable activity name."""
+    aapt_binary_path = os.path.join(
+        environment.get_platform_resources_directory(), 'aapt')
+    aapt_command = '%s dump badging %s' % (aapt_binary_path, apk_path)
+    output = android.adb.execute_command(aapt_command, timeout=60)
+    if not output:
+      return None
+    match = re.search(r"launchable-activity: name='([^']+)'", output)
+    if match:
+      return match.group(1)
+    return None
+
+  def _get_instrumentation_runner(self, apk_path):
+    """Return the instrumentation runner."""
+    aapt_binary_path = os.path.join(
+        environment.get_platform_resources_directory(), 'aapt')
+    aapt_command = '%s dump xmltree %s AndroidManifest.xml' % (aapt_binary_path,
+                                                               apk_path)
+    output = android.adb.execute_command(aapt_command, timeout=60)
+    if not output:
+      return None
+
+    lines = output.splitlines()
+    found_instrumentation = False
+    for line in lines:
+      if 'E: instrumentation' in line:
+        found_instrumentation = True
+        continue
+      if found_instrumentation and 'android:name' in line:
+        match = re.search(r'="([^"]+)"', line)
+        if match:
+          return match.group(1)
+    return None
+
+  def _get_device_corpus_paths(self, corpus_directories):
+    return [self._get_device_path(path) for path in corpus_directories]
+
+  def _get_device_path(self, local_path):
+    root_directory = environment.get_root_directory()
+    return os.path.join(android.constants.DEVICE_FUZZING_DIR,
+                        os.path.relpath(local_path, root_directory))
+
+  def _copy_local_directories_to_device(self, local_directories):
+    for local_directory in sorted(set(local_directories)):
+      self.copy_local_directory_to_device(local_directory)
+
+  def copy_local_directory_to_device(self, local_directory):
+    device_directory = self._get_device_path(local_directory)
+    android.adb.remove_directory(device_directory, recreate=True)
+    android.adb.copy_local_directory_to_remote(local_directory,
+                                               device_directory)
+
+  def _copy_local_directories_from_device(self, local_directories):
+    for local_directory in sorted(set(local_directories)):
+      device_directory = self._get_device_path(local_directory)
+      shell.remove_directory(local_directory, recreate=True)
+      android.adb.copy_remote_directory_to_local(device_directory,
+                                                 local_directory)
+
+  def fuzz(self,
+           corpus_directories,
+           fuzz_timeout,
+           artifact_prefix=None,
+           additional_args=None,
+           extra_env=None):
+    sync_directories = list(corpus_directories)
+    if artifact_prefix:
+      sync_directories.append(artifact_prefix)
+
+    self._copy_local_directories_to_device(sync_directories)
+    device_corpus_dirs = self._get_device_corpus_paths(corpus_directories)
+
+    if artifact_prefix:
+      artifact_prefix = self._get_device_path(artifact_prefix)
+
+    fuzzer_args = []
+    if additional_args:
+      fuzzer_args.extend(additional_args)
+    fuzzer_args.extend(device_corpus_dirs)
+    fuzzer_args_str = ' '.join(fuzzer_args)
+
+    if self.instrumentation_runner:
+      device_cache_dir = f'/data/user/0/{self.package_name}/cache'
+      android.adb.run_shell_command(
+          f'mkdir -p {device_cache_dir}', log_output=True)
+      android.adb.run_shell_command(
+          f'chmod 777 {device_cache_dir}', log_output=True)
+      device_stdout_file = os.path.join(device_cache_dir, 'fuzzer_output.txt')
+      args = [
+          'shell',
+          'am',
+          'instrument',
+          '-w',
+          '-e',
+          'org.chromium.native_test.NativeUnitTestActivity',
+          'org.chromium.native_test.NativeUnitTestActivity',
+          '-e',
+          f'{self.instrumentation_runner}.NativeTestActivity',
+          'org.chromium.native_test.NativeUnitTestActivity',
+          '-e',
+          'org.chromium.native_test.NativeTestInstrumentationTestRunner.'
+          'NativeTestActivity',
+          'org.chromium.native_test.NativeUnitTestActivity',
+          '-e',
+          'org.chromium.native_test.NativeTestInstrumentationTestRunner.'
+          'StdoutFile',
+          device_stdout_file,
+          '-e',
+          f'{self.instrumentation_runner}.StdoutFile',
+          device_stdout_file,
+      ]
+
+      if self.library_under_test:
+        args.extend([
+            '-e',
+            'org.chromium.native_test.NativeTestInstrumentationTestRunner.'
+            'LibraryUnderTest', self.library_under_test
+        ])
+      if self.auxiliary_libraries:
+        args.extend([
+            '-e',
+            'org.chromium.native_test.NativeTestInstrumentationTestRunner.'
+            'AuxiliaryLibraries', self.auxiliary_libraries
+        ])
+
+      args.extend([
+          '-e',
+          'org.chromium.native_test.NativeTest.CommandLineFlags',
+          f'"{fuzzer_args_str}"',
+      ])
+
+      args.append(f'{self.package_name}/{self.instrumentation_runner}')
+      logs.info(f'Starting Instrumentation: {self.package_name}/'
+                f'{self.instrumentation_runner} with args: {fuzzer_args_str}')
+    elif self.launchable_activity:
+      device_stdout_file = None
+      args = [
+          'shell', 'am', 'start', '-n',
+          f'{self.package_name}/{self.launchable_activity}', '-e',
+          'org.chromium.native_test.NativeTest.CommandLineFlags',
+          f'"{fuzzer_args_str}"'
+      ]
+      logs.info(f'Starting APK: {self.package_name}/{self.launchable_activity} '
+                f'with args: {fuzzer_args_str}')
+    else:
+      raise LibFuzzerError('No launchable activity or instrumentation found.')
+
+    # Workaround for wrap.sh execution on Android.
+    # We write a custom wrap.sh and extract ASan RT to the app's secure
+    # app_native_libs directory to avoid linker restrictions, and run with
+    # SELinux permissive.
+    target_dir = f"/data/data/{self.package_name}/app_native_libs"
+    wrap_sh_path = f"{target_dir}/wrap.sh"
+    selinux_revert = False
+
+    try:
+      pm_path_output = android.adb.run_shell_command(
+          f'pm path {self.package_name}')
+      if pm_path_output and pm_path_output.startswith('package:'):
+        apk_path = pm_path_output.split(':')[1].strip()
+
+        # Ensure target directory exists and is writable
+        android.adb.run_shell_command(f'su 0 mkdir -p {target_dir}')
+        android.adb.run_shell_command(f'su 0 chmod 777 {target_dir}')
+
+        # Clean up old files if any
+        android.adb.run_shell_command(f'su 0 rm -f {wrap_sh_path}')
+        android.adb.run_shell_command(
+            f'su 0 rm -f {target_dir}/libclang_rt.asan-*.so')
+
+        # Extract ASan runtime using unzip directly to target_dir
+        unzip_asan_cmd = (f'su 0 unzip -o -j {apk_path} '
+                          f'"lib/x86_64/libclang_rt.asan-*.so" -d {target_dir}')
+        unzip_asan_result = android.adb.run_shell_command(unzip_asan_cmd)
+        logs.info(f"DEBUG: unzip ASan rt result: {unzip_asan_result.strip()}")
+
+        # Generate custom wrap.sh content (ensuring LF line endings and marker)
+        wrap_sh_content = (
+            f"#!/system/bin/sh\n"
+            f"HERE=\"{target_dir}\"\n"
+            f"_ASAN_OPTIONS=\"log_to_syslog=false,"
+            f"allow_user_segv_handler=1\"\n"
+            f"_ASAN_OPTIONS=\"$_ASAN_OPTIONS,"
+            f"strict_memcmp=0,use_sigaltstack=1\"\n"
+            f"_LD_PRELOAD=\"$HERE/libclang_rt.asan-x86_64-android.so\"\n"
+            f"echo wrap_sh_ran > /data/local/tmp/wrap_marker.txt\n"
+            f"log -t cr_wrap.sh -- \"Launching with ASAN enabled.\"\n"
+            f"log -t cr_wrap.sh -- \"Command: $0 $@\"\n"
+            f"log -t cr_wrap.sh -- \"LD_PRELOAD=$_LD_PRELOAD\"\n"
+            f"log -t cr_wrap.sh -- \"ASAN_OPTIONS=$_ASAN_OPTIONS\"\n"
+            f"export LD_PRELOAD=\"$_LD_PRELOAD\"\n"
+            f"export ASAN_OPTIONS=\"$_ASAN_OPTIONS\"\n"
+            f"exec \"$@\"\n")
+
+        # Write wrap.sh locally and push to avoid permission issues
+        import tempfile
+        with tempfile.NamedTemporaryFile(
+            mode='w', suffix='.sh', delete=False) as temp_wrap:
+          temp_wrap.write(wrap_sh_content)
+          temp_wrap_path = temp_wrap.name
+
+        tmp_wrap_path = "/data/local/tmp/temp_wrap.sh"
+        android.adb.copy_local_file_to_remote(temp_wrap_path, tmp_wrap_path)
+        os.remove(temp_wrap_path)
+
+        android.adb.run_shell_command(f"su 0 cp {tmp_wrap_path} {wrap_sh_path}")
+        android.adb.run_shell_command(f"su 0 rm -f {tmp_wrap_path}")
+
+        if android.adb.file_exists(wrap_sh_path):
+          android.adb.run_shell_command(f'su 0 chmod 777 {wrap_sh_path}')
+          android.adb.run_shell_command(
+              f'su 0 chmod 777 {target_dir}/libclang_rt.asan-*.so')
+
+          selinux_status = android.adb.run_shell_command('su 0 getenforce')
+          logs.info(f"DEBUG: Initial SELinux status: {selinux_status.strip()}")
+          if selinux_status.strip() == 'Enforcing':
+            android.adb.run_shell_command('su 0 setenforce 0')
+            selinux_revert = True
+            logs.info('DEBUG: Disabled SELinux (set to Permissive) '
+                      'for wrap.sh execution.')
+
+          logs.info(f"DEBUG: Setting wrap property to {wrap_sh_path}")
+          setprop_result = android.adb.run_shell_command(
+              f'su 0 setprop wrap.{self.package_name} {wrap_sh_path}')
+          if setprop_result:
+            logs.warning(
+                f"DEBUG: setprop output/error: {setprop_result.strip()}")
+        else:
+          logs.error(f"DEBUG: Failed to create wrap.sh at {wrap_sh_path}")
+          wrap_sh_path = None
+      else:
+        logs.error(f"DEBUG: Could not get APK path for {self.package_name}")
+        wrap_sh_path = None
+    except Exception as e:
+      logs.error(f"DEBUG: Error setting up wrap.sh workaround: {e}")
+      wrap_sh_path = None
+
+    try:
+      result = self.run_and_wait(
+          additional_args=args,
+          timeout=self.get_total_timeout(fuzz_timeout),
+          max_stdout_len=MAX_OUTPUT_LEN)
+    finally:
+      # Clear the wrapper property.
+      android.adb.run_shell_command(f'su 0 setprop wrap.{self.package_name} ""')
+      # Restore SELinux if we changed it
+      if selinux_revert:
+        android.adb.run_shell_command('su 0 setenforce 1')
+        logs.info("DEBUG: Restored SELinux to Enforcing.")
+
+      # Check for marker
+      marker_path = "/data/local/tmp/wrap_marker.txt"
+      if android.adb.file_exists(marker_path):
+        logs.info("DEBUG: SUCCESS! wrap.sh executed (marker found).")
+        marker_content = android.adb.run_shell_command(
+            f"su 0 cat {marker_path}")
+        logs.info(f"DEBUG: marker content: {marker_content.strip()}")
+        android.adb.run_shell_command(f"su 0 rm -f {marker_path}")
+      else:
+        logs.error(
+            "DEBUG: FAILURE! wrap.sh did NOT execute (marker not found).")
+
+      # Clean up wrap.sh and ASan runtime
+      if wrap_sh_path:
+        android.adb.run_shell_command(f'su 0 rm -f {wrap_sh_path}')
+        android.adb.run_shell_command(
+            f'su 0 rm -f {target_dir}/libclang_rt.asan-*.so')
+
+    logs.info(f'DEBUG: adb command run: {result.command}')
+    logs.info(f'DEBUG: adb command return code: {result.return_code}')
+    logs.info(f'DEBUG: adb command output: {result.output}')
+
+    if device_stdout_file:
+      fuzzer_output = android.adb.run_shell_command(f'cat {device_stdout_file}')
+      android.adb.remove_file(device_stdout_file)
+      if fuzzer_output:
+        logs.info(f'Fuzzer native stdout:\n{fuzzer_output}')
+        result.output = f'{fuzzer_output}\n\n{result.output}'
+
+    result.output = (
+        f'{result.output}\n\nLogcat:\n{android.logger.log_output()}')
+    logs.info(f'Fuzzer run output for {self.package_name}:\n{result.output}')
+    self._copy_local_directories_from_device(sync_directories)
+    return result
+
+
 def get_runner(fuzzer_path, temp_dir=None, use_minijail=None):
   """Get a libfuzzer runner."""
   if use_minijail is None:
@@ -1190,7 +1636,30 @@ def get_runner(fuzzer_path, temp_dir=None, use_minijail=None):
       raise undercoat.UndercoatError('Instance handle not provided.')
     runner = FuchsiaUndercoatLibFuzzerRunner(fuzzer_path, instance_handle)
   elif is_android:
-    runner = AndroidLibFuzzerRunner(fuzzer_path, build_dir)
+    # Find the APK dynamically under build_dir.
+    # We first look for a fuzzer-specific APK (e.g. {fuzzer_name}-debug.apk
+    # or {fuzzer_name}.apk).
+    fuzzer_name = os.path.basename(fuzzer_path)
+    base_apk_path = None
+
+    specific_apk_names = [f'{fuzzer_name}-debug.apk', f'{fuzzer_name}.apk']
+    for root, _, files in os.walk(build_dir):
+      # Check for specific APK first
+      for apk_name in specific_apk_names:
+        if apk_name in files:
+          base_apk_path = os.path.join(root, apk_name)
+          break
+      if base_apk_path:
+        break
+
+    if base_apk_path:
+      logs.info(f'Using Android APK runner for {fuzzer_name}. '
+                f'APK path: {base_apk_path}')
+      runner = AndroidApkLibFuzzerRunner(base_apk_path, build_dir)
+    else:
+      logs.info(f'Using Android command-line binary runner for {fuzzer_name}. '
+                f'Path: {fuzzer_path}')
+      runner = AndroidLibFuzzerRunner(fuzzer_path, build_dir)
   else:
     runner = LibFuzzerRunner(fuzzer_path, cwd=cwd)
 
diff --git a/src/clusterfuzz/_internal/bot/fuzzers/utils.py b/src/clusterfuzz/_internal/bot/fuzzers/utils.py
index a7494c0ce86..19f92bec1de 100644
--- a/src/clusterfuzz/_internal/bot/fuzzers/utils.py
+++ b/src/clusterfuzz/_internal/bot/fuzzers/utils.py
@@ -26,7 +26,7 @@
 from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.system import shell
 
-ALLOWED_FUZZ_TARGET_EXTENSIONS = ['', '.exe', '.par']
+ALLOWED_FUZZ_TARGET_EXTENSIONS = ['', '.exe', '.par', '.apk']
 FUZZ_TARGET_SEARCH_BYTES = [b'LLVMFuzzerTestOneInput', b'LLVMFuzzerRunDriver']
 VALID_TARGET_NAME_REGEX = re.compile(r'^[a-zA-Z0-9@_.-]+$')
 BLOCKLISTED_TARGET_NAME_REGEX = re.compile(r'^(jazzer_driver.*|jazzerjs)$')
diff --git a/src/clusterfuzz/_internal/bot/init_scripts/android.py b/src/clusterfuzz/_internal/bot/init_scripts/android.py
index 64293ee76a0..acfc8c0ca43 100644
--- a/src/clusterfuzz/_internal/bot/init_scripts/android.py
+++ b/src/clusterfuzz/_internal/bot/init_scripts/android.py
@@ -53,5 +53,9 @@ def run():
   # Wait until battery charges to a minimum level and temperature threshold.
   android.battery.wait_until_good_state()
 
+  # Ensure adb runs as root for the rest of the session (essential for
+  # engine fuzzers).
+  android.adb.run_as_root()
+
   # Initialize environment settings.
   android.device.initialize_environment()
diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
index 1936a8e3ba6..e7893afc6e6 100644
--- a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
+++ b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
@@ -2299,6 +2299,28 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env):
     fuzz_task_input.global_blacklisted_functions.extend(
         leak_blacklist.get_global_blacklisted_functions())
 
+  # Inject custom binary details if this is a custom binary job.
+  if uworker_env.get('CUSTOM_BINARY'):
+    job = data_types.Job.query(data_types.Job.name == job_type).get()
+    if job and job.custom_binary_key:
+      uworker_env['CUSTOM_BINARY_KEY'] = job.custom_binary_key
+      uworker_env['CUSTOM_BINARY_FILENAME'] = job.custom_binary_filename
+      uworker_env['CUSTOM_BINARY_REVISION'] = str(job.custom_binary_revision)
+
+      # Generate signed URL for the custom binary to avoid 403 on uworker
+      try:
+        custom_builds_bucket = local_config.ProjectConfig().get(
+            'custom_builds.bucket')
+        if custom_builds_bucket:
+          gcs_path = f'/{custom_builds_bucket}/{job.custom_binary_key}'
+        else:
+          gcs_path = f'/{storage.blobs_bucket()}/{job.custom_binary_key}'
+
+        uworker_env['CUSTOM_BINARY_SIGNED_URL'] = (
+            storage.get_signed_download_url(gcs_path))
+      except Exception as e:
+        logs.error(f'Failed to generate signed URL for custom binary: {e}')
+
   return uworker_msg_pb2.Input(  # pylint: disable=no-member
       fuzz_task_input=fuzz_task_input,
       job_type=job_type,
diff --git a/src/clusterfuzz/_internal/build_management/build_manager.py b/src/clusterfuzz/_internal/build_management/build_manager.py
index 02b84e4d18f..1d25c024be6 100644
--- a/src/clusterfuzz/_internal/build_management/build_manager.py
+++ b/src/clusterfuzz/_internal/build_management/build_manager.py
@@ -893,11 +893,16 @@ def setup(self):
 class CustomBuild(Build):
   """Custom binary."""
 
-  def __init__(self, base_build_dir, custom_binary_key, custom_binary_filename,
-               custom_binary_revision):
+  def __init__(self,
+               base_build_dir,
+               custom_binary_key,
+               custom_binary_filename,
+               custom_binary_revision,
+               custom_binary_signed_url=None):
     super().__init__(base_build_dir, custom_binary_revision)
     self.custom_binary_key = custom_binary_key
     self.custom_binary_filename = custom_binary_filename
+    self.custom_binary_signed_url = custom_binary_signed_url
     self._build_dir = os.path.join(self.base_build_dir, 'custom')
 
   @property
@@ -913,20 +918,35 @@ def _unpack_custom_build(self):
 
     build_local_archive = os.path.join(self.build_dir,
                                        self.custom_binary_filename)
-    custom_builds_bucket = local_config.ProjectConfig().get(
-        'custom_builds.bucket')
 
     download_start_time = time.time()
 
-    if custom_builds_bucket:
-      directory = os.path.dirname(build_local_archive)
-      if not os.path.exists(directory):
-        os.makedirs(directory)
-      gcs_path = f'/{custom_builds_bucket}/{self.custom_binary_key}'
-      storage.copy_file_from(gcs_path, build_local_archive)
-    elif not blobs.read_blob_to_disk(self.custom_binary_key,
-                                     build_local_archive):
-      return False
+    # Try to download using signed URL if available (uworker friendly)
+    if self.custom_binary_signed_url:
+      logs.info('Downloading custom binary using signed URL.')
+      try:
+        storage.download_signed_url_to_file(self.custom_binary_signed_url,
+                                            build_local_archive)
+        if not os.path.exists(build_local_archive) or os.path.getsize(
+            build_local_archive) == 0:
+          logs.error('Downloaded custom binary is empty or missing.')
+          return False
+      except Exception as e:
+        logs.error(f'Failed to download custom binary using signed URL: {e}')
+        return False
+    else:
+      custom_builds_bucket = local_config.ProjectConfig().get(
+          'custom_builds.bucket')
+
+      if custom_builds_bucket:
+        directory = os.path.dirname(build_local_archive)
+        if not os.path.exists(directory):
+          os.makedirs(directory)
+        gcs_path = f'/{custom_builds_bucket}/{self.custom_binary_key}'
+        storage.copy_file_from(gcs_path, build_local_archive)
+      elif not blobs.read_blob_to_disk(self.custom_binary_key,
+                                       build_local_archive):
+        return False
 
     _emit_job_build_retrieval_metric(download_start_time, 'download',
                                      self._build_type)
@@ -1416,7 +1436,25 @@ def setup_symbolized_builds(revision):
 def setup_custom_binary():
   """Set up the custom binary for a particular job."""
   job_name = environment.get_value('JOB_NAME')
-  # Verify that this is really a custom binary job.
+
+  # Try to get from environment first (to avoid Datastore query on uworker)
+  custom_binary_key = environment.get_value('CUSTOM_BINARY_KEY')
+  custom_binary_filename = environment.get_value('CUSTOM_BINARY_FILENAME')
+  custom_binary_revision = environment.get_value('CUSTOM_BINARY_REVISION')
+  custom_binary_signed_url = environment.get_value('CUSTOM_BINARY_SIGNED_URL')
+
+  if custom_binary_key and custom_binary_filename:
+    logs.info('Using custom binary details from environment.')
+    base_build_dir = _base_build_dir('')
+    build = CustomBuild(
+        base_build_dir, custom_binary_key, custom_binary_filename,
+        int(custom_binary_revision or 0), custom_binary_signed_url)
+    if build.setup():
+      return build
+    return None
+
+  # Fallback to Datastore query
+  logs.info('Custom binary details not in environment, querying Datastore...')
   job = data_types.Job.query(data_types.Job.name == job_name).get()
   if not job or not job.custom_binary_key or not job.custom_binary_filename:
     logs.error(
diff --git a/src/clusterfuzz/_internal/datastore/data_types.py b/src/clusterfuzz/_internal/datastore/data_types.py
index e648a58bea3..24ad730341e 100644
--- a/src/clusterfuzz/_internal/datastore/data_types.py
+++ b/src/clusterfuzz/_internal/datastore/data_types.py
@@ -883,6 +883,10 @@ class Config(Model):
   # functional bugs.
   relax_security_bug_restrictions = ndb.BooleanProperty(default=False)
 
+  # TODO(b/422775458) - Clean up after completing the migration to V4
+  # Flag to use the V4 Android Build API instead of V3.
+  use_android_build_api_v4 = ndb.BooleanProperty(default=False)
+
   # Coverage reports bucket.
   coverage_reports_bucket = ndb.StringProperty(default='')
 
diff --git a/src/clusterfuzz/_internal/platforms/android/adb.py b/src/clusterfuzz/_internal/platforms/android/adb.py
index 7d3c712206b..e5f62dee6f2 100755
--- a/src/clusterfuzz/_internal/platforms/android/adb.py
+++ b/src/clusterfuzz/_internal/platforms/android/adb.py
@@ -225,7 +225,12 @@ def get_adb_path():
 
 def get_device_state():
   """Return the device status."""
-  if environment.is_android_emulator():
+  # Emulators do not support the fastboot protocol and cannot enter physical
+  # ramdump mode. The original check was backwards (checking emulator instead
+  # of physical devices), causing a 20-second fastboot timeout on every state
+  # check on emulators. We correct this to only run fastboot checks on physical
+  # devices.
+  if not environment.is_android_emulator():
     fastboot_state = run_fastboot_command(
         ['getvar', 'is-ramdump-mode'], timeout=GET_DEVICE_STATE_TIMEOUT)
     if fastboot_state and 'is-ramdump-mode: yes' in fastboot_state:
@@ -600,7 +605,7 @@ def _get_device_path_for_usb():
         open('/sys/bus/usb/devices/%s/devnum' % device_id).read().strip())
     return '/dev/bus/usb/%03d/%03d' % (bus_number, device_number)
 
-  if environment.is_android_cuttlefish():
+  if environment.is_android_cuttlefish() or environment.is_android_emulator():
     return None
 
   device_serial = environment.get_value('ANDROID_SERIAL')
@@ -612,7 +617,7 @@ def _get_device_path_for_usb():
 
 def reset_usb():
   """Reset USB bus for a device serial."""
-  if environment.is_android_cuttlefish():
+  if environment.is_android_cuttlefish() or environment.is_android_emulator():
     # Nothing to do here.
     return True
 
diff --git a/src/clusterfuzz/_internal/platforms/android/app.py b/src/clusterfuzz/_internal/platforms/android/app.py
index f90a0d0cb95..396782f5e9e 100644
--- a/src/clusterfuzz/_internal/platforms/android/app.py
+++ b/src/clusterfuzz/_internal/platforms/android/app.py
@@ -93,9 +93,25 @@ def get_package_name(apk_path=None):
   return match.group(1)
 
 
-def install(package_apk_path):
-  """Install a package from an apk path."""
-  return adb.run_command(['install', '-r', package_apk_path])
+def install(package_apk_path: str, **kwargs):
+  """Install a package from an apk path.
+
+  Args:
+    package_apk_path: Path to the apk file to install.
+    **kwargs: Additional arguments to pass to the install command.
+  """
+  cmd = ['install', '-r']
+  for key, value in kwargs.items():
+    if not value:
+      continue
+
+    flag = '-' + key if len(key) == 1 else '--' + key.replace('_', '-')
+    cmd.append(flag)
+    if not isinstance(value, bool):
+      cmd.append(str(value))
+
+  cmd.append(package_apk_path)
+  return adb.run_command(cmd)
 
 
 def is_installed(package_name):
diff --git a/src/clusterfuzz/_internal/platforms/android/device.py b/src/clusterfuzz/_internal/platforms/android/device.py
index 074aa569b47..34ca44171ff 100755
--- a/src/clusterfuzz/_internal/platforms/android/device.py
+++ b/src/clusterfuzz/_internal/platforms/android/device.py
@@ -352,6 +352,29 @@ def initialize_environment():
   environment.set_value('LOG_TASK_TIMES', True)
 
 
+def _needs_no_streaming_for_asan() -> bool:
+  """
+  Asan setup requires non-streaming installation because streaming
+  installation prevents wrap.sh from being invoked on modern Android
+  versions which is necessary for ASan to function correctly.
+
+  Returns:
+    bool: True if the job or device requires ASan, False otherwise.
+  """
+  is_asan_job = False
+  job_name = environment.get_value('JOB_NAME')
+  if job_name:
+    memory_tool = environment.get_memory_tool_name(job_name)
+    if memory_tool == 'ASAN':
+      is_asan_job = True
+
+  is_asan_device = (
+      environment.get_value('ASAN_DEVICE_SETUP') or
+      settings.get_sanitizer_tool_name() == 'asan')
+
+  return is_asan_job or is_asan_device
+
+
 def install_application_if_needed(apk_path, force_update):
   """Install application package if it does not exist on device
   or if force_update is set."""
@@ -377,7 +400,7 @@ def install_application_if_needed(apk_path, force_update):
   # package list or force_update flag has been set.
   if force_update or not app.is_installed(package_name):
     app.uninstall(package_name)
-    app.install(apk_path)
+    app.install(apk_path, no_streaming=_needs_no_streaming_for_asan())
 
     if not app.is_installed(package_name):
       logs.error('Package %s was not installed successfully.' % package_name)
diff --git a/src/clusterfuzz/_internal/platforms/android/fetch_artifact.py b/src/clusterfuzz/_internal/platforms/android/fetch_artifact.py
index 4aaf7f40aad..e86cced06c6 100644
--- a/src/clusterfuzz/_internal/platforms/android/fetch_artifact.py
+++ b/src/clusterfuzz/_internal/platforms/android/fetch_artifact.py
@@ -48,6 +48,22 @@
     "gs://android-haiku/target-cuttlefish/stable_build_info.json")
 
 
+def _use_v4():
+  """Return True if we should use V4 Android Build API."""
+  try:
+    use_v4 = db_config.get_value('use_android_build_api_v4') or False
+    logs.info(
+        'AndroidBuildAPI feature flag status read.',
+        use_android_build_api_v4=use_v4)
+    return use_v4
+  except Exception as e:
+    logs.error(
+        'AndroidBuildAPI error reading feature flag use_android_build_api_v4. '
+        'Defaulting to False.',
+        error=str(e))
+    return False
+
+
 def execute_request_with_retries(request):
   """Executes request and retries on failure."""
   result = None
@@ -68,24 +84,68 @@ def download_artifact(client, bid, target, attempt_id, name, output_directory,
   logs.info('artifact to download: %s' % name)
   logs.info('output_directory: %s' % output_directory)
   logs.info('output_filename: %s' % output_filename)
-  artifact_query = client.buildartifact().get(
-      buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
+
+  version_tag = 'V4' if _use_v4() else 'V3'
+  logs.info(
+      'AndroidBuildAPI download_artifact started.',
+      api_version=version_tag,
+      operation='download_artifact',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      artifact_name=name)
+
+  if _use_v4():
+    artifact_query = client.buildartifacts().get(
+        buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
+  else:
+    artifact_query = client.buildartifact().get(
+        buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
   artifact = execute_request_with_retries(artifact_query)
   if artifact is None:
-    logs.error(f'Artifact unreachable with name {name}, target {target} '
-               f'and build id {bid}.')
+    logs.error(
+        'AndroidBuildAPI download_artifact failed: artifact metadata '
+        'unreachable.',
+        api_version=version_tag,
+        operation='download_artifact',
+        build_id=bid,
+        target=target,
+        attempt_id=attempt_id,
+        artifact_name=name,
+        status='failed')
     return None
 
   # Lucky us, we always have the size.
   size = int(artifact['size'])
+  logs.info(
+      'AndroidBuildAPI download_artifact metadata retrieved successfully.',
+      api_version=version_tag,
+      operation='download_artifact',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      artifact_name=name,
+      size=size)
 
   chunksize = -1
   if size >= DEFAULT_CHUNK_SIZE:
     chunksize = DEFAULT_CHUNK_SIZE
 
   # Just like get, except get_media.
-  dl_request = client.buildartifact().get_media(
-      buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
+  logs.info(
+      'AndroidBuildAPI download_artifact media download started.',
+      api_version=version_tag,
+      operation='download_artifact_media',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      artifact_name=name)
+  if _use_v4():
+    dl_request = client.buildartifacts().get_media(
+        buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
+  else:
+    dl_request = client.buildartifact().get_media(
+        buildId=bid, target=target, attemptId=attempt_id, resourceId=name)
 
   if output_filename:
     file_name = output_filename
@@ -95,7 +155,16 @@ def download_artifact(client, bid, target, attempt_id, name, output_directory,
   output_path = os.path.join(output_directory, file_name)
   # If the artifact already exists, then bail out.
   if os.path.exists(output_path) and os.path.getsize(output_path) == size:
-    logs.info('Artifact %s already exists, skipping download.' % name)
+    logs.info(
+        'AndroidBuildAPI download_artifact skipped (file already exists).',
+        api_version=version_tag,
+        operation='download_artifact',
+        build_id=bid,
+        target=target,
+        attempt_id=attempt_id,
+        artifact_name=name,
+        output_path=output_path,
+        status='skipped_exists')
     return output_path
 
   logs.info('Downloading artifact %s.' % name)
@@ -118,6 +187,16 @@ def download_artifact(client, bid, target, attempt_id, name, output_directory,
           percent_completed = (size_completed * 100.0) / size
           logs.info('%.1f%% complete.' % percent_completed)
 
+  logs.info(
+      'AndroidBuildAPI download_artifact completed successfully.',
+      api_version=version_tag,
+      operation='download_artifact',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      artifact_name=name,
+      output_path=output_path,
+      status='success')
   return output_path
 
 
@@ -127,21 +206,44 @@ def get_artifacts_for_build(client,
                             attempt_id: str = 'latest',
                             regexp: Optional[str] = None) -> List[str]:
   """Return list of artifacts for a given build."""
-  if not regexp:
-    request = client.buildartifact().list(
-        buildId=bid, target=target, attemptId=attempt_id)
+  version_tag = 'V4' if _use_v4() else 'V3'
+  logs.info(
+      'AndroidBuildAPI get_artifacts_for_build started.',
+      api_version=version_tag,
+      operation='get_artifacts_for_build',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      regexp=regexp)
+
+  if _use_v4():
+    if not regexp:
+      request = client.buildartifacts().list(
+          buildId=bid, target=target, attemptId=attempt_id)
+    else:
+      request = client.buildartifacts().list(
+          buildId=bid,
+          target=target,
+          attemptId=attempt_id,
+          nameRegexp=regexp,
+          maxResults=100)
   else:
-    request = client.buildartifact().list(
-        buildId=bid,
-        target=target,
-        attemptId=attempt_id,
-        nameRegexp=regexp,
-        maxResults=100)
+    if not regexp:
+      request = client.buildartifact().list(
+          buildId=bid, target=target, attemptId=attempt_id)
+    else:
+      request = client.buildartifact().list(
+          buildId=bid,
+          target=target,
+          attemptId=attempt_id,
+          nameRegexp=regexp,
+          maxResults=100)
 
   request_str = (f'{request.uri}, {request.method}, '
                  f'{request.body}, {request.methodId}')
 
   artifacts = []
+
   results = []
   while request:
     result = execute_request_with_retries(request)
@@ -151,7 +253,21 @@ def get_artifacts_for_build(client,
     if result and 'artifacts' in result:
       for artifact in result['artifacts']:
         artifacts.append(artifact)
-    request = client.buildartifact().list_next(request, result)
+    if _use_v4():
+      request = client.buildartifacts().list_next(request, result)
+    else:
+      request = client.buildartifact().list_next(request, result)
+
+  logs.info(
+      'AndroidBuildAPI get_artifacts_for_build completed.',
+      api_version=version_tag,
+      operation='get_artifacts_for_build',
+      build_id=bid,
+      target=target,
+      attempt_id=attempt_id,
+      regexp=regexp,
+      artifacts_count=len(artifacts),
+      status='success' if artifacts else 'empty')
 
   if not artifacts:
     logs.error(f'No artifact found for target {target}, build id {bid}.\n'
@@ -174,11 +290,24 @@ def get_client():
   credentials = ServiceAccountCredentials.from_json_keyfile_dict(
       json.loads(build_apiary_service_account_private_key),
       scopes='https://www.googleapis.com/auth/androidbuild.internal')
-  client = apiclient.discovery.build(
-      'androidbuildinternal',
-      'v3',
-      credentials=credentials,
-      static_discovery=False)
+  if _use_v4():
+    logs.info(
+        'AndroidBuildAPI client initialization started.', api_version='V4')
+    client = apiclient.discovery.build(
+        'androidbuildinternal',
+        'v4',
+        discoveryServiceUrl=
+        'https://androidbuild-pa.googleapis.com/$discovery/rest?version=v4',
+        credentials=credentials,
+        static_discovery=False)
+  else:
+    logs.info(
+        'AndroidBuildAPI client initialization started.', api_version='V3')
+    client = apiclient.discovery.build(
+        'androidbuildinternal',
+        'v3',
+        credentials=credentials,
+        static_discovery=False)
 
   return client
 
@@ -218,25 +347,59 @@ def get_latest_artifact_info(branch, target, signed=False, stable_build=False):
     if 'bid' in build_info and build_info['bid'] != '0':
       return build_info
 
-  request = client.build().list(  # pylint: disable=no-member
-      buildType='submitted',
+  version_tag = 'V4' if _use_v4() else 'V3'
+  logs.info(
+      'AndroidBuildAPI get_latest_artifact_info started.',
+      api_version=version_tag,
+      operation='get_latest_artifact_info',
       branch=branch,
       target=target,
-      successful=True,
-      maxResults=1,
       signed=signed)
+  if _use_v4():
+    request = client.builds().list(  # pylint: disable=no-member
+        buildType='submitted',
+        branch=branch,
+        target=target,
+        successful=True,
+        maxResults=1,
+        signed=signed)
+  else:
+    request = client.build().list(  # pylint: disable=no-member
+        buildType='submitted',
+        branch=branch,
+        target=target,
+        successful=True,
+        maxResults=1,
+        signed=signed)
   request_str = (f'{request.uri}, {request.method}, '
                  f'{request.body}, {request.methodId}')
 
   builds = execute_request_with_retries(request)
   if not builds:
-    logs.error(f'No build found for target {target}, branch {branch}, '
-               f'request: {request_str}.')
+    logs.error(
+        'AndroidBuildAPI get_latest_artifact_info failed: no builds found.',
+        api_version=version_tag,
+        operation='get_latest_artifact_info',
+        branch=branch,
+        target=target,
+        signed=signed,
+        status='failed',
+        request_str=request_str)
     return None
 
   build = builds['builds'][0]
   bid = build['buildId']
   target = build['target']['name']
+
+  logs.info(
+      'AndroidBuildAPI get_latest_artifact_info completed.',
+      api_version=version_tag,
+      operation='get_latest_artifact_info',
+      branch=branch,
+      target=target,
+      signed=signed,
+      build_id=bid,
+      status='success')
   return {'bid': bid, 'branch': branch, 'target': target}
 
 
diff --git a/src/clusterfuzz/_internal/platforms/android/symbols_downloader.py b/src/clusterfuzz/_internal/platforms/android/symbols_downloader.py
index 112d86fc215..0ba650c8b73 100644
--- a/src/clusterfuzz/_internal/platforms/android/symbols_downloader.py
+++ b/src/clusterfuzz/_internal/platforms/android/symbols_downloader.py
@@ -35,6 +35,11 @@ def get_repo_prop_archive_filename(build_id, target):
 
 def should_download_symbols():
   """Return True if we should continue to download symbols."""
+  # Uworkers do not have Datastore access to retrieve the API key needed for
+  # downloading symbols.
+  if environment.is_uworker():
+    return False
+
   # For local testing, we do not have access to the cloud storage bucket with
   # the symbols. In this case, just bail out. We have archived symbols for
   # google builds only.
@@ -192,7 +197,20 @@ def download_trusty_symbols_if_needed(symbols_directory, app_name, bid):
   elif 'oriole' in device or 'raven' in device or 'bluejay' in device:
     ab_target = 'slider-fuzz-test-debug'
   else:
-    logs.error(f'Unsupported device {device}.')
+    # Emulators (like sdk_gdesktop) do not run the physical Trusty Secure OS
+    # TEE. Therefore, no official Trusty symbols exist for them at the moment
+    # in GCS. Instead of logging a confusing ERROR that can be mistaken for a
+    # fuzz task failure, we log a verbose INFO message and return early.
+    if environment.is_android_emulator(
+    ) or 'sdk_gdesktop' in device or 'emulator' in device:
+      logs.info(
+          f'Skipping Trusty symbols download for emulator device "{device}". '
+          'Virtual/emulator devices do not run a physical Trusty '
+          'Trusted Execution Environment (TEE), '
+          'so no official Trusty symbol archives are available or needed.')
+    else:
+      logs.error(f'Unsupported device {device}.')
+    return
 
   branch = 'polygon-trusty-whitechapel-master'
   if not bid:
diff --git a/src/clusterfuzz/_internal/tests/core/platforms/android/app_test.py b/src/clusterfuzz/_internal/tests/core/platforms/android/app_test.py
index 9f5e4f54693..7095042c150 100644
--- a/src/clusterfuzz/_internal/tests/core/platforms/android/app_test.py
+++ b/src/clusterfuzz/_internal/tests/core/platforms/android/app_test.py
@@ -14,10 +14,13 @@
 """Tests for app functions."""
 
 import os
+from unittest import mock
+from unittest import TestCase
 
 from clusterfuzz._internal.platforms.android import app
 from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.tests.test_libs import android_helpers
+from clusterfuzz._internal.tests.test_libs import helpers
 
 
 class IsInstalledTest(android_helpers.AndroidTest):
@@ -61,3 +64,33 @@ def test_apk_path_in_arg(self):
     """Test apk path passed as argument."""
     self.assertEqual(
         app.get_package_name(self.test_apk_path), self.test_apk_pkg_name)
+
+
+class InstallTest(TestCase):
+  """Tests install."""
+
+  def setUp(self):
+    super().setUp()
+    helpers.patch_environ(self)
+    self.run_command_patcher = mock.patch(
+        'clusterfuzz._internal.platforms.android.adb.run_command')
+    self.mock_run_command = self.run_command_patcher.start()
+    self.addCleanup(self.run_command_patcher.stop)
+
+  def test_install_normal(self):
+    """Test normal installation without any additional flags."""
+    app.install('/path/to/app.apk')
+    self.mock_run_command.assert_called_once_with(
+        ['install', '-r', '/path/to/app.apk'])
+
+  def test_install_with_additional_flags(self):
+    """Test installation with additional flags."""
+    app.install('/path/to/app.apk', g=True, t=True)
+    self.mock_run_command.assert_called_once_with(
+        ['install', '-r', '-g', '-t', '/path/to/app.apk'])
+
+  def test_install_with_valued_flags(self):
+    """Test installation with flags that take string/numeric values."""
+    app.install('/path/to/app.apk', abi='x86', no_streaming=True)
+    self.mock_run_command.assert_called_once_with(
+        ['install', '-r', '--abi', 'x86', '--no-streaming', '/path/to/app.apk'])
diff --git a/src/clusterfuzz/_internal/tests/core/platforms/android/device_test.py b/src/clusterfuzz/_internal/tests/core/platforms/android/device_test.py
index 9c6eef0b715..b79c3a36be5 100644
--- a/src/clusterfuzz/_internal/tests/core/platforms/android/device_test.py
+++ b/src/clusterfuzz/_internal/tests/core/platforms/android/device_test.py
@@ -16,6 +16,7 @@
 import unittest
 
 from clusterfuzz._internal.platforms.android import device
+from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.tests.test_libs import android_helpers
 from clusterfuzz._internal.tests.test_libs import helpers
 
@@ -81,3 +82,41 @@ def test_uworker_bypass(self):
     self.mock.is_uworker.return_value = True
     device.add_test_accounts_if_needed()
     self.mock.get_value.assert_not_called()
+
+
+class NeedsNoStreamingForAsanTest(unittest.TestCase):
+  """Tests _needs_no_streaming_for_asan."""
+
+  # pylint: disable=protected-access
+
+  def setUp(self):
+    super().setUp()
+    helpers.patch_environ(self)
+    helpers.patch(self, [
+        'clusterfuzz._internal.platforms.android.settings.get_sanitizer_tool_name',
+    ])
+    self.mock.get_sanitizer_tool_name.return_value = None
+
+  def test_normal_job(self):
+    """Test outcome for a normal job."""
+    self.assertFalse(device._needs_no_streaming_for_asan())
+
+  def test_asan_job(self):
+    """Test outcome for an ASan job."""
+    environment.set_value('JOB_NAME', 'android_asan_job')
+    self.assertTrue(device._needs_no_streaming_for_asan())
+
+  def test_hwasan_job(self):
+    """Test outcome for a non ASAN job like HWASan."""
+    environment.set_value('JOB_NAME', 'android_hwasan_job')
+    self.assertFalse(device._needs_no_streaming_for_asan())
+
+  def test_asan_device_env(self):
+    """Test outcome for an ASan device via ASAN_DEVICE_SETUP env."""
+    environment.set_value('ASAN_DEVICE_SETUP', True)
+    self.assertTrue(device._needs_no_streaming_for_asan())
+
+  def test_asan_device_flavor(self):
+    """Test outcome for an ASan device via settings build flavor."""
+    self.mock.get_sanitizer_tool_name.return_value = 'asan'
+    self.assertTrue(device._needs_no_streaming_for_asan())