From 474e56ef8e9faa0c544ec61c5f0dd4bab4e887aa Mon Sep 17 00:00:00 2001
From: somew1nd <wfq559@outlook.com>
Date: Sat, 22 Aug 2026 19:43:52 -0700
Subject: [PATCH] feat: support orientation-aware recording and device
 wakefulness

Roll scrcpy recordings by device rotation and duration, and publish timeline manifests for playback and analysis.

Add seamless segmented playback in the showcase UI and configurable stay-awake strategies for Android automation.
---
 .env.example                                  |   3 +
 apps/admin_console/routers/media.py           |   2 +
 apps/admin_console/services/media_service.py  |  30 ++
 .../floating-video-player.component.html      |   2 +-
 .../floating-video-player.component.scss      |   3 +-
 .../floating-video-player.component.ts        |  73 +++-
 .../src/app/services/agent.service.ts         |  21 +-
 artemis/clients/ui_automator_client.py        | 170 ++++++++
 artemis/controllers/android_controller.py     |  60 +--
 artemis/controllers/unified_controller.py     | 395 ++++++++----------
 artemis/drivers/android/adb_driver.py         |  17 +-
 artemis/utils/video.py                        | 331 +++++++++++++++
 tests/unit/test_ui_automator_client_awake.py  | 165 ++++++++
 tests/unit/test_unified_controller_video.py   | 188 ++++++++-
 14 files changed, 1150 insertions(+), 310 deletions(-)
 create mode 100644 tests/unit/test_ui_automator_client_awake.py

diff --git a/.env.example b/.env.example
index 27a9fc5..2c6b072 100644
--- a/.env.example
+++ b/.env.example
@@ -19,6 +19,9 @@ OCR_API_KEY=
 ADB_HOST=127.0.0.1
 ADB_PORT=5037
 # ADB_DEVICE_SERIAL=
+# Keep the display awake, falling back to a wake lock or safe heartbeat when needed.
+# Set to false if Artemis is used with a personal device whose screen should time out.
+ARTEMIS_KEEP_DEVICE_AWAKE=true
 
 # Artemis Execution Defaults (flash or pro)
 ARTEMIS_DEFAULT_PROFILE=flash
diff --git a/apps/admin_console/routers/media.py b/apps/admin_console/routers/media.py
index ae40caf..4603ed6 100644
--- a/apps/admin_console/routers/media.py
+++ b/apps/admin_console/routers/media.py
@@ -92,10 +92,12 @@ async def get_session_video(session_id: str):
     row = session_repo.get_session_by_id(session_id)
     row_dict = dict(row) if row else {"session_id": session_id}
     v_url = media_service.resolve_video_url(row_dict, video_rec_map, video_idx)
+    video_segments = media_service.resolve_video_segments(v_url)
     return {
         "session_id": session_id,
         "has_video": bool(v_url),
         "video_url": v_url,
+        "video_segments": video_segments,
     }
 
 
diff --git a/apps/admin_console/services/media_service.py b/apps/admin_console/services/media_service.py
index 2e566de..fda93d5 100644
--- a/apps/admin_console/services/media_service.py
+++ b/apps/admin_console/services/media_service.py
@@ -127,6 +127,36 @@ class MediaService:
 
         return v_url
 
+    @classmethod
+    def resolve_video_segments(cls, video_url: str | None) -> list[dict[str, Any]]:
+        """Resolve a recording sidecar manifest into browser-safe segment URLs."""
+        if not video_url or not video_url.startswith("/videos/"):
+            return []
+        relative = urllib.parse.unquote(video_url.removeprefix("/videos/"))
+        video_path = (WORKSPACE_ROOT / relative).resolve()
+        manifest_path = video_path.parent / "recording.json"
+        if not manifest_path.is_file():
+            return []
+        try:
+            payload = json.loads(manifest_path.read_text(encoding="utf-8"))
+            segments = []
+            for item in payload.get("segments", []):
+                segment_path = (manifest_path.parent / str(item["file"])).resolve()
+                if segment_path.parent != manifest_path.parent.resolve() or not segment_path.is_file():
+                    continue
+                segments.append(
+                    {
+                        "url": cls.path_to_video_url(segment_path),
+                        "start": float(item.get("start", 0)),
+                        "duration": float(item.get("duration", 0)),
+                        "width": int(item.get("width", 0)),
+                        "height": int(item.get("height", 0)),
+                    }
+                )
+            return segments
+        except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError):
+            return []
+
     @staticmethod
     def unwrap_payload(obj: Any) -> Any:
         """Unwraps trace payloads, storing base64 images as disk files and
diff --git a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.html b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.html
index 2e883fc..4cadb08 100644
--- a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.html
+++ b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.html
@@ -134,7 +134,7 @@
         <div class="video-container" *ngIf="!videoLoadError()">
           <video 
             #videoRef
-            [src]="videoUrl" 
+            [src]="currentVideoUrl()"
             playsinline
             (loadedmetadata)="onLoadedMetadata()"
             (timeupdate)="onTimeUpdate()"
diff --git a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.scss b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.scss
index bd15f33..0c54dd6 100644
--- a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.scss
+++ b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.scss
@@ -373,16 +373,15 @@
     .video-container {
       position: relative;
       width: 100%;
-      height: 100%;
       display: flex;
       flex-direction: column;
       align-items: center;
       justify-content: center;
       background: #ffffff;
-      min-height: 220px;
 
       video {
         width: 100%;
+        height: auto;
         max-height: 520px;
         display: block;
         object-fit: contain;
diff --git a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts
index 5a0724c..536e66d 100644
--- a/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts
+++ b/apps/showcase_ui/src/app/components/floating-video-player/floating-video-player.component.ts
@@ -51,6 +51,13 @@ export class FloatingVideoPlayerComponent {
   public videoLoadError = signal<boolean>(false);
   public liveStreamUrl = signal<string>('/api/stream/device-live');
   public liveStreamError = signal<boolean>(false);
+  public activeSegmentIndex = signal<number>(0);
+  public currentVideoUrl = computed(() => {
+    const segments = this.agentService.activeVideoSegments();
+    return segments[this.activeSegmentIndex()]?.url || this.agentService.activeVideoUrl();
+  });
+  private pendingLocalTime: number | null = null;
+  private pendingAutoplay = false;
 
   // Available speed options
   public speedOptions = [0.5, 1.0, 1.5, 2.0, 4.0];
@@ -74,9 +81,11 @@ export class FloatingVideoPlayerComponent {
     effect(
       () => {
         const url = this.agentService.activeVideoUrl();
+        const segments = this.agentService.activeVideoSegments();
         this.videoLoadError.set(false);
+        this.activeSegmentIndex.set(0);
         this.currentTime.set(0);
-        this.duration.set(0);
+        this.duration.set(segments.reduce((sum, segment) => sum + segment.duration, 0));
         this.isPlaying.set(false);
       },
       { allowSignalWrites: true }
@@ -177,18 +186,32 @@ export class FloatingVideoPlayerComponent {
   public onLoadedMetadata(): void {
     const v = this.videoRef?.nativeElement;
     if (v) {
-      this.duration.set(v.duration || 0);
+      const segments = this.agentService.activeVideoSegments();
+      if (!segments.length) {
+        this.duration.set(v.duration || 0);
+      }
       this.videoLoadError.set(false);
       v.playbackRate = this.playbackRate();
-      v.loop = this.isLooping();
+      v.loop = this.isLooping() && !segments.length;
+      v.muted = this.isMuted();
+      if (this.pendingLocalTime !== null) {
+        v.currentTime = Math.max(0, Math.min(v.duration || this.pendingLocalTime, this.pendingLocalTime));
+        this.pendingLocalTime = null;
+      }
+      if (this.pendingAutoplay) {
+        this.pendingAutoplay = false;
+        v.play().catch(() => {});
+      }
     }
   }
 
   public onTimeUpdate(): void {
     const v = this.videoRef?.nativeElement;
     if (v) {
-      this.currentTime.set(v.currentTime);
-      if (v.duration && v.duration !== this.duration()) {
+      const segments = this.agentService.activeVideoSegments();
+      const segment = segments[this.activeSegmentIndex()];
+      this.currentTime.set((segment?.start || 0) + v.currentTime);
+      if (!segments.length && v.duration && v.duration !== this.duration()) {
         this.duration.set(v.duration);
       }
     }
@@ -203,6 +226,17 @@ export class FloatingVideoPlayerComponent {
   }
 
   public onEnded(): void {
+    const segments = this.agentService.activeVideoSegments();
+    if (segments.length && this.activeSegmentIndex() < segments.length - 1) {
+      this.pendingLocalTime = 0;
+      this.pendingAutoplay = true;
+      this.activeSegmentIndex.update(index => index + 1);
+      return;
+    }
+    if (segments.length && this.isLooping()) {
+      this.seek(0, true);
+      return;
+    }
     this.isPlaying.set(false);
   }
 
@@ -225,16 +259,31 @@ export class FloatingVideoPlayerComponent {
   }
 
   public restart(): void {
-    const v = this.videoRef?.nativeElement;
-    if (!v) return;
-    v.currentTime = 0;
-    v.play().catch(() => {});
+    this.seek(0, true);
   }
 
-  public seek(seconds: number): void {
+  public seek(seconds: number, autoplay = false): void {
     const v = this.videoRef?.nativeElement;
     if (!v) return;
-    v.currentTime = Math.max(0, Math.min(this.duration(), seconds));
+    const target = Math.max(0, Math.min(this.duration(), seconds));
+    const segments = this.agentService.activeVideoSegments();
+    if (!segments.length) {
+      v.currentTime = target;
+      if (autoplay) v.play().catch(() => {});
+      return;
+    }
+    let index = segments.findIndex(segment => target < segment.start + segment.duration);
+    if (index < 0) index = segments.length - 1;
+    const localTime = Math.max(0, target - segments[index].start);
+    if (index === this.activeSegmentIndex()) {
+      v.currentTime = localTime;
+      if (autoplay) v.play().catch(() => {});
+    } else {
+      this.pendingLocalTime = localTime;
+      this.pendingAutoplay = autoplay || !v.paused;
+      this.activeSegmentIndex.set(index);
+    }
+    this.currentTime.set(target);
   }
 
   public onScrub(event: Event): void {
@@ -256,7 +305,7 @@ export class FloatingVideoPlayerComponent {
     this.isLooping.set(next);
     const v = this.videoRef?.nativeElement;
     if (v) {
-      v.loop = next;
+      v.loop = next && !this.agentService.activeVideoSegments().length;
     }
   }
 
diff --git a/apps/showcase_ui/src/app/services/agent.service.ts b/apps/showcase_ui/src/app/services/agent.service.ts
index 129801a..7432f27 100644
--- a/apps/showcase_ui/src/app/services/agent.service.ts
+++ b/apps/showcase_ui/src/app/services/agent.service.ts
@@ -22,6 +22,14 @@ import { Session, ModelInfo, TaskQueueItem, AgentStatusResponse } from '../core/
 import { StepItemData } from '../core/models/stream.model';
 export type { Session, ModelInfo, TaskQueueItem, AgentStatusResponse, StepItemData };
 
+export interface VideoSegment {
+  url: string;
+  start: number;
+  duration: number;
+  width: number;
+  height: number;
+}
+
 @Injectable({
   providedIn: 'root'
 })
@@ -104,6 +112,7 @@ export class AgentService {
   public isVideoWindowOpen = signal<boolean>(false);
   public isVideoMinimized = signal<boolean>(false);
   public activeVideoUrl = signal<string | null>(null);
+  public activeVideoSegments = signal<VideoSegment[]>([]);
   public activeVideoTitle = signal<string>('');
   public isVideoLoading = signal<boolean>(false);
 
@@ -1000,29 +1009,33 @@ export class AgentService {
     this.isVideoWindowOpen.set(true);
     this.isVideoMinimized.set(false);
 
-    if (targetUrl) {
-      this.activeVideoUrl.set(targetUrl);
-    } else if (targetSessionId) {
+    this.activeVideoUrl.set(targetUrl);
+    this.activeVideoSegments.set([]);
+    if (targetSessionId) {
       this.isVideoLoading.set(true);
-      this.http.get<{ session_id: string; has_video: boolean; video_url: string | null }>(`/api/sessions/${targetSessionId}/video`).subscribe({
+      this.http.get<{ session_id: string; has_video: boolean; video_url: string | null; video_segments?: VideoSegment[] }>(`/api/sessions/${targetSessionId}/video`).subscribe({
         next: (res) => {
           this.isVideoLoading.set(false);
           if (res && res.has_video && res.video_url) {
             this.activeVideoUrl.set(res.video_url);
+            this.activeVideoSegments.set(res.video_segments || []);
             this.rawSessions.update((list) =>
               list.map((s) => s.session_id === targetSessionId ? { ...s, video_url: res.video_url || undefined } : s)
             );
           } else {
             this.activeVideoUrl.set(null);
+            this.activeVideoSegments.set([]);
           }
         },
         error: () => {
           this.isVideoLoading.set(false);
           this.activeVideoUrl.set(null);
+          this.activeVideoSegments.set([]);
         }
       });
     } else {
       this.activeVideoUrl.set(null);
+      this.activeVideoSegments.set([]);
     }
   }
 
diff --git a/artemis/clients/ui_automator_client.py b/artemis/clients/ui_automator_client.py
index 6d92507..f40ece3 100644
--- a/artemis/clients/ui_automator_client.py
+++ b/artemis/clients/ui_automator_client.py
@@ -21,8 +21,10 @@ Handles Maestro blocker detection and removal before connecting.
 
 import base64
 from io import BytesIO
+import os
 import re as _re
 import subprocess
+import threading
 import time
 from typing import TYPE_CHECKING
 import xml.etree.ElementTree as ET
@@ -40,6 +42,129 @@ logger = get_logger(__name__)
 
 
 MAESTRO_PACKAGE = "dev.mobile.maestro"
+AWAKE_STRATEGY_STAY_ON = "stay_on"
+AWAKE_STRATEGY_WAKE_LOCK = "wake_lock"
+AWAKE_STRATEGY_HEARTBEAT = "heartbeat"
+AWAKE_HEARTBEAT_INTERVAL_SECONDS = 5.0
+
+
+def _run_awake_adb_command(
+    device_id: str, args: list[str], description: str
+) -> subprocess.CompletedProcess[str] | None:
+    """Run one non-fatal ADB command used by the awake strategy."""
+    try:
+        result = subprocess.run(
+            ["adb", "-s", device_id, *args],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+        if result.returncode != 0:
+            detail = (result.stderr or result.stdout).strip()
+            logger.warning(
+                f"Could not {description} on {device_id}: {detail or 'ADB command failed'}"
+            )
+        return result
+    except Exception as e:
+        logger.warning(f"Could not {description} on {device_id}: {e}")
+        return None
+
+
+def _keep_device_awake(device_id: str) -> str | None:
+    """Best-effort preparation of an Android device for UI automation.
+
+    ``svc power stayon true`` asks Android to keep the display on while the
+    device has any external power source.  It avoids changing the user's
+    normal battery-powered screen timeout.  The wake and dismiss commands
+    also recover a device that went to sleep before Artemis connected;
+    ``wm dismiss-keyguard`` cannot bypass a secure PIN/password keyguard.
+
+    The primary strategy is verified through PowerManager. If it is not
+    actually active, Artemis tries a shell-owned bright-screen wake lock. If
+    that command is unavailable, the caller starts a periodic KEYCODE_WAKEUP
+    heartbeat. Neither fallback changes the user's screen-timeout setting.
+
+    This must never make device connection fail. Some vendor Android builds
+    restrict one or more shell commands, so failures are logged only.
+    """
+    enabled = os.environ.get("ARTEMIS_KEEP_DEVICE_AWAKE", "true").strip().lower()
+    if enabled in {"0", "false", "no", "off"}:
+        return None
+
+    commands = (
+        ("enable stay-awake while powered", ["shell", "svc", "power", "stayon", "true"]),
+        ("wake display", ["shell", "input", "keyevent", "KEYCODE_WAKEUP"]),
+        ("dismiss non-secure keyguard", ["shell", "wm", "dismiss-keyguard"]),
+    )
+    for description, args in commands:
+        _run_awake_adb_command(device_id, args, description)
+
+    power_state = _run_awake_adb_command(
+        device_id, ["shell", "dumpsys", "power"], "verify stay-awake state"
+    )
+    if (
+        power_state is not None
+        and power_state.returncode == 0
+        and _re.search(r"\bmStayOn=true\b", power_state.stdout)
+    ):
+        logger.info(f"Stay-awake primary strategy verified on {device_id}")
+        return AWAKE_STRATEGY_STAY_ON
+
+    logger.warning(
+        f"Stay-awake primary strategy is not active on {device_id}; "
+        "trying a screen-bright wake lock"
+    )
+    _run_awake_adb_command(
+        device_id,
+        [
+            "shell",
+            "cmd",
+            "power",
+            "set-wakelock",
+            "acquire",
+            "-d",
+            "0",
+            "SCREEN_BRIGHT_WAKE_LOCK",
+        ],
+        "acquire a screen-bright wake lock",
+    )
+    wake_locks = _run_awake_adb_command(
+        device_id,
+        ["shell", "cmd", "power", "set-wakelock", "list"],
+        "verify the screen-bright wake lock",
+    )
+    if (
+        wake_locks is not None
+        and wake_locks.returncode == 0
+        and _re.search(r"SCREEN_BRIGHT_WAKE_LOCK:.*\bheld=true\b", wake_locks.stdout)
+    ):
+        logger.info(f"Screen-bright wake lock verified on {device_id}")
+        return AWAKE_STRATEGY_WAKE_LOCK
+
+    logger.warning(
+        f"Screen-bright wake lock is unavailable on {device_id}; "
+        f"using a {AWAKE_HEARTBEAT_INTERVAL_SECONDS:g}-second wakeup heartbeat"
+    )
+    return AWAKE_STRATEGY_HEARTBEAT
+
+
+def _release_screen_wake_lock(device_id: str) -> None:
+    """Release the shell-owned screen wake lock acquired for this session."""
+    _run_awake_adb_command(
+        device_id,
+        [
+            "shell",
+            "cmd",
+            "power",
+            "set-wakelock",
+            "release",
+            "-d",
+            "0",
+            "SCREEN_BRIGHT_WAKE_LOCK",
+        ],
+        "release the screen-bright wake lock",
+    )
 
 
 class _CyFunctionDetectorMeta(type):
@@ -257,6 +382,41 @@ class UIAutomatorClient:
         """
         self._device_id = device_id
         self._device: Device | None = None
+        self._awake_strategy: str | None = None
+        self._heartbeat_stop_event: threading.Event | None = None
+        self._heartbeat_thread: threading.Thread | None = None
+
+    def _start_awake_heartbeat(self) -> None:
+        """Start a daemon heartbeat that safely refreshes Android user activity."""
+        if self._heartbeat_thread is not None and self._heartbeat_thread.is_alive():
+            return
+
+        stop_event = threading.Event()
+        self._heartbeat_stop_event = stop_event
+
+        def heartbeat() -> None:
+            while not stop_event.wait(AWAKE_HEARTBEAT_INTERVAL_SECONDS):
+                _run_awake_adb_command(
+                    self._device_id,
+                    ["shell", "input", "keyevent", "KEYCODE_WAKEUP"],
+                    "send the stay-awake heartbeat",
+                )
+
+        self._heartbeat_thread = threading.Thread(
+            target=heartbeat,
+            name=f"artemis-awake-{self._device_id}",
+            daemon=True,
+        )
+        self._heartbeat_thread.start()
+
+    def _stop_awake_heartbeat(self) -> None:
+        """Stop the session heartbeat without delaying shutdown."""
+        if self._heartbeat_stop_event is not None:
+            self._heartbeat_stop_event.set()
+        if self._heartbeat_thread is not None:
+            self._heartbeat_thread.join(timeout=2.0)
+        self._heartbeat_stop_event = None
+        self._heartbeat_thread = None
 
     def _ensure_connected(self) -> "Device":
         """Ensure connection to the device, handling Maestro blocker.
@@ -276,6 +436,12 @@ class UIAutomatorClient:
         # Ensure Maestro is not blocking us
         _ensure_maestro_not_installed(self._device_id)
 
+        # Configure this automation session before UIAutomator2 reads the screen.
+        if self._awake_strategy is None:
+            self._awake_strategy = _keep_device_awake(self._device_id)
+            if self._awake_strategy == AWAKE_STRATEGY_HEARTBEAT:
+                self._start_awake_heartbeat()
+
         # Connect to device
         logger.info(f"Connecting UIAutomator2 to device: {self._device_id}")
         self._device = u2.connect(self._device_id)
@@ -400,6 +566,10 @@ class UIAutomatorClient:
 
     def disconnect(self) -> None:
         """Disconnect from the device."""
+        self._stop_awake_heartbeat()
+        if self._awake_strategy == AWAKE_STRATEGY_WAKE_LOCK:
+            _release_screen_wake_lock(self._device_id)
+        self._awake_strategy = None
         self._device = None
         logger.info("UIAutomator2 client disconnected")
 
diff --git a/artemis/controllers/android_controller.py b/artemis/controllers/android_controller.py
index 53b0395..3c7f1df 100644
--- a/artemis/controllers/android_controller.py
+++ b/artemis/controllers/android_controller.py
@@ -45,10 +45,12 @@ from artemis.utils.video import (
     DEFAULT_MAX_DURATION_SECONDS,
     RecordingSession,
     VideoRecordingResult,
+    build_scrcpy_record_command,
     cleanup_video_segments,
     concatenate_videos,
     get_active_session,
     has_active_session,
+    normalize_recording_to_mp4,
     remove_active_session,
     set_active_session,
     trim_video,
@@ -561,6 +563,8 @@ class AndroidDeviceController(MobileDeviceController):
                 start_time=time.time(),
                 data_engine_start_time=self.data_engine_start_time,
                 local_video_path=local_video_path,
+                capture_width=self.device_width,
+                capture_height=self.device_height,
             )
 
             # Persist to local database if Data Engine is active
@@ -578,18 +582,9 @@ class AndroidDeviceController(MobileDeviceController):
                     logger.error(f"Failed to persist video recording start to DB: {db_err}")
 
             # Start scrcpy in background
-            cmd = [
-                "scrcpy",
-                "--serial",
-                self.device_id,
-                "--no-window",
-                "--record",
-                str(local_video_path),
-                "--record-format",
-                "mkv",
-                "--video-bit-rate",
-                "2M",
-            ]
+            cmd = build_scrcpy_record_command(
+                "scrcpy", self.device_id, local_video_path, lock_capture_orientation=False
+            )
 
             process = await asyncio.create_subprocess_exec(
                 *cmd,
@@ -760,18 +755,9 @@ class AndroidDeviceController(MobileDeviceController):
                     output_dir = mkv_path.parent
                     new_video_path = output_dir / f"recording_{session.android_segment_index}.mkv"
 
-                    cmd = [
-                        "scrcpy",
-                        "--serial",
-                        self.device_id,
-                        "--no-window",
-                        "--record",
-                        str(new_video_path),
-                        "--record-format",
-                        "mkv",
-                        "--video-bit-rate",
-                        "2M",
-                    ]
+                    cmd = build_scrcpy_record_command(
+                        "scrcpy", self.device_id, new_video_path, lock_capture_orientation=False
+                    )
 
                     process = await asyncio.create_subprocess_exec(
                         *cmd,
@@ -892,21 +878,11 @@ class AndroidDeviceController(MobileDeviceController):
             )
 
     async def _convert_mkv_to_mp4(self, mkv_path: Path, mp4_path: Path) -> bool:
-        """Convert MKV to MP4 using ffmpeg."""
-        try:
-            process = await asyncio.create_subprocess_exec(
-                "ffmpeg",
-                "-y",
-                "-i",
-                str(mkv_path),
-                "-c",
-                "copy",
-                str(mp4_path),
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-            await process.wait()
-            return mp4_path.exists()
-        except Exception as e:
-            logger.error(f"Failed to convert MKV to MP4: {e}")
-            return False
+        """Normalize MKV into a fixed-size, browser-safe MP4."""
+        session = get_active_session(self.device_id)
+        return await normalize_recording_to_mp4(
+            mkv_path,
+            mp4_path,
+            session.capture_width if session else self.device_width,
+            session.capture_height if session else self.device_height,
+        )
diff --git a/artemis/controllers/unified_controller.py b/artemis/controllers/unified_controller.py
index 0fe38ef..ebf2e8e 100644
--- a/artemis/controllers/unified_controller.py
+++ b/artemis/controllers/unified_controller.py
@@ -15,6 +15,8 @@
 import asyncio
 import os
 from pathlib import Path
+import signal
+import subprocess
 import tempfile
 import time
 from typing import Any
@@ -33,17 +35,23 @@ from artemis.controllers.types import (
 )
 from artemis.utils.logger import get_logger
 from artemis.utils.video import (
+    ANDROID_RECORDING_SEGMENT_SECONDS,
     DEFAULT_MAX_DURATION_SECONDS,
     RecordingSession,
     VideoRecordingResult,
+    build_scrcpy_record_command,
     cleanup_video_segments,
     concatenate_videos,
+    get_android_display_state,
     get_active_session,
-    get_ffmpeg_path,
     has_active_session,
+    normalize_recording_to_mp4,
+    remux_recording_to_mp4,
+    render_timeline_clip,
     remove_active_session,
     set_active_session,
     trim_video,
+    write_recording_manifest,
 )
 
 logger = get_logger(__name__)
@@ -63,6 +71,29 @@ class UnifiedMobileController:
     def controller(self) -> Any:
         return self._driver
 
+    @staticmethod
+    async def _spawn_scrcpy(command: list[str]) -> asyncio.subprocess.Process:
+        kwargs: dict[str, Any] = {
+            "stdout": asyncio.subprocess.PIPE,
+            "stderr": asyncio.subprocess.PIPE,
+        }
+        if os.name == "nt":
+            kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+        return await asyncio.create_subprocess_exec(*command, **kwargs)
+
+    @staticmethod
+    async def _stop_scrcpy(process: asyncio.subprocess.Process) -> None:
+        """Ask scrcpy to flush its recorder before falling back to termination."""
+        if process.returncode is not None:
+            return
+        try:
+            process.send_signal(signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGINT)
+            await asyncio.wait_for(process.wait(), timeout=8.0)
+        except (ProcessLookupError, TimeoutError):
+            if process.returncode is None:
+                process.terminate()
+                await asyncio.wait_for(process.wait(), timeout=5.0)
+
     async def tap_at(
         self,
         x: int,
@@ -301,68 +332,6 @@ class UnifiedMobileController:
                     message="Recording file not found",
                 )
 
-            # Check if scrcpy process terminated and restart if needed (synchronous fallback)
-            crashed = False
-            restart_message = ""
-            if session.process and session.process.returncode is not None:
-                logger.warning(f"scrcpy process for {device_id} has terminated unexpectedly")
-                crashed = True
-                if mkv_path.exists() and mkv_path not in session.android_video_segments:
-                    session.android_video_segments.append(mkv_path)
-                try:
-                    session.android_segment_index += 1
-                    output_dir = mkv_path.parent
-                    new_video_path = output_dir / f"recording_{session.android_segment_index}.mkv"
-                    cmd = [
-                        "scrcpy",
-                        "--serial",
-                        device_id,
-                        "--no-window",
-                        "--record",
-                        str(new_video_path),
-                        "--record-format",
-                        "mkv",
-                        "--video-bit-rate",
-                        "2M",
-                    ]
-                    process = await asyncio.create_subprocess_exec(
-                        *cmd,
-                        stdout=asyncio.subprocess.PIPE,
-                        stderr=asyncio.subprocess.PIPE,
-                    )
-                    await asyncio.sleep(0.8)
-                    if process.returncode is None:
-                        session.process = process
-                        session.local_video_path = new_video_path
-                        mkv_path = new_video_path
-                        restart_message = "Successfully restarted scrcpy."
-                    else:
-                        stderr = await process.stderr.read()
-                        restart_message = f"Failed to restart scrcpy: {stderr.decode()}"
-                except Exception as e:
-                    restart_message = f"Failed to restart scrcpy: {e}"
-
-            # Determine source file to trim from (support multi-segment seamless trimming)
-            all_segments = [p for p in session.android_video_segments if p.exists()]
-            if mkv_path.exists() and mkv_path not in all_segments:
-                all_segments.append(mkv_path)
-
-            if not all_segments:
-                return VideoRecordingResult(
-                    success=False,
-                    message="No recording segments found on disk",
-                )
-
-            temp_combined_file = None
-            if len(all_segments) > 1:
-                temp_combined_file = mkv_path.parent / "temp_combined_active.mkv"
-                concat_ok = await concatenate_videos(all_segments, temp_combined_file)
-                trim_source = (
-                    temp_combined_file if (concat_ok and temp_combined_file.exists()) else mkv_path
-                )
-            else:
-                trim_source = all_segments[0]
-
             current_time = time.time()
             video_duration = current_time - session.start_time
 
@@ -415,8 +384,18 @@ class UnifiedMobileController:
 
             success = False
             for attempt in range(3):
-                success = await trim_video(
-                    trim_source,
+                timeline_segments = [dict(record) for record in session.android_segment_records]
+                active_start = session.android_segment_started_at or session.start_time
+                if mkv_path.exists():
+                    timeline_segments.append(
+                        {
+                            "path": mkv_path,
+                            "start": max(0.0, active_start - session.start_time),
+                            "end": max(0.0, current_time - session.start_time),
+                        }
+                    )
+                success = await render_timeline_clip(
+                    timeline_segments,
                     video_start_relative_time,
                     video_end_relative_time,
                     trim_output_path,
@@ -426,13 +405,6 @@ class UnifiedMobileController:
                 logger.warning(f"ffmpeg trim attempt {attempt + 1} failed, retrying in 0.5s...")
                 await asyncio.sleep(0.5)
 
-            # Cleanup temp combined MKV if created
-            if temp_combined_file and temp_combined_file.exists():
-                try:
-                    temp_combined_file.unlink()
-                except Exception:
-                    pass
-
             if not success or not trim_output_path.exists():
                 return VideoRecordingResult(
                     success=False,
@@ -445,11 +417,6 @@ class UnifiedMobileController:
             duration = video_end_relative_time - video_start_relative_time
 
             message = f"Video segment retrieved for range {actual_start:.1f}s to {actual_end:.1f}s"
-            if crashed:
-                message += (
-                    f". Warning: scrcpy crashed. Data loss may have occurred. {restart_message}"
-                )
-
             res = VideoRecordingResult(
                 success=True,
                 message=message,
@@ -469,61 +436,98 @@ class UnifiedMobileController:
                 message=f"Failed to get video segment: {e}",
             )
 
-    async def _recording_watchdog(self, device_id: str) -> None:
-        """Background watchdog to automatically detect crashes and recover scrcpy recording."""
-        while True:
-            session = get_active_session(device_id)
-            if not session or not session.is_active:
-                break
-            proc = session.process
-            if proc is None:
-                break
+    @staticmethod
+    def _segment_mp4_path(source_path: Path, index: int) -> Path:
+        return source_path.parent / ("recording.mp4" if index == 0 else f"recording_{index:03d}.mp4")
+
+    @staticmethod
+    async def _remux_segment_record(record: dict[str, Any]) -> bool:
+        source_path = Path(record["path"])
+        output_path = Path(record["output_path"])
+        success = await remux_recording_to_mp4(source_path, output_path)
+        if success:
+            record["path"] = output_path
             try:
-                ret = await proc.wait()
+                source_path.unlink()
+            except OSError:
+                pass
+        return success
+
+    def _finalize_current_segment(self, session: RecordingSession, end_time: float) -> None:
+        source_path = session.local_video_path
+        if not source_path or not source_path.exists():
+            return
+        if any(Path(record["path"]) == source_path for record in session.android_segment_records):
+            return
+        start_time = session.android_segment_started_at or session.start_time
+        output_path = self._segment_mp4_path(source_path, session.android_segment_index)
+        session.android_video_segments.append(source_path)
+        session.android_segment_records.append(
+            {
+                "path": source_path,
+                "output_path": output_path,
+                "start": max(0.0, start_time - session.start_time),
+                "end": max(0.0, end_time - session.start_time),
+                "rotation": session.android_rotation,
+                "conversion_scheduled": True,
+            }
+        )
+        session.android_conversion_tasks.append(
+            asyncio.create_task(self._remux_segment_record(session.android_segment_records[-1]))
+        )
+
+    async def _start_next_recording_segment(
+        self, session: RecordingSession, display_state: tuple[int, int, int] | None
+    ) -> bool:
+        session.android_segment_index += 1
+        output_dir = session.local_video_path.parent
+        new_video_path = output_dir / f"recording_{session.android_segment_index:03d}.mkv"
+        process = await self._spawn_scrcpy(
+            build_scrcpy_record_command("scrcpy", session.device_id, new_video_path)
+        )
+        await asyncio.sleep(0.8)
+        if process.returncode is not None:
+            stderr = await process.stderr.read()
+            logger.error(f"Failed to start scrcpy segment: {stderr.decode(errors='replace')}")
+            return False
+        session.process = process
+        session.local_video_path = new_video_path
+        session.android_segment_started_at = time.time()
+        if display_state:
+            session.android_rotation, session.capture_width, session.capture_height = display_state
+        return True
+
+    async def _recording_watchdog(self, device_id: str) -> None:
+        """Roll fixed-orientation segments and recover scrcpy crashes."""
+        try:
+            while True:
+                await asyncio.sleep(0.5)
                 session = get_active_session(device_id)
-                if not session or not session.is_active:
-                    break
-
-                logger.warning(
-                    f"scrcpy recording process for {device_id} terminated unexpectedly (code {ret}). "
-                    f"Auto-restarting recording segment {session.android_segment_index + 1}..."
+                if not session or not session.is_active or not session.process:
+                    return
+                now = time.time()
+                display_state = await get_android_display_state(device_id)
+                rotated = bool(
+                    display_state
+                    and session.android_rotation is not None
+                    and display_state[0] != session.android_rotation
                 )
-                if session.local_video_path and session.local_video_path.exists():
-                    if session.local_video_path not in session.android_video_segments:
-                        session.android_video_segments.append(session.local_video_path)
+                segment_age = now - (session.android_segment_started_at or session.start_time)
+                crashed = session.process.returncode is not None
+                if not crashed and not rotated and segment_age < ANDROID_RECORDING_SEGMENT_SECONDS:
+                    continue
 
-                session.android_segment_index += 1
-                output_dir = session.local_video_path.parent
-                new_video_path = output_dir / f"recording_{session.android_segment_index}.mkv"
-                cmd = [
-                    "scrcpy",
-                    "--serial",
-                    device_id,
-                    "--no-window",
-                    "--record",
-                    str(new_video_path),
-                    "--record-format",
-                    "mkv",
-                    "--video-bit-rate",
-                    "2M",
-                ]
-                new_proc = await asyncio.create_subprocess_exec(
-                    *cmd,
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
-                )
-                session.process = new_proc
-                session.local_video_path = new_video_path
-                await asyncio.sleep(0.8)
-                if new_proc.returncode is not None:
-                    stderr = await new_proc.stderr.read()
-                    logger.error(f"Failed to auto-restart scrcpy recording: {stderr.decode()}")
-                    break
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.error(f"Error in recording watchdog for {device_id}: {e}")
-                break
+                if not crashed:
+                    await self._stop_scrcpy(session.process)
+                self._finalize_current_segment(session, time.time())
+                reason = "rotation" if rotated else "time limit" if not crashed else "recorder crash"
+                logger.info(f"Rolling screen recording segment after {reason}")
+                if not await self._start_next_recording_segment(session, display_state):
+                    return
+        except asyncio.CancelledError:
+            return
+        except Exception as e:
+            logger.error(f"Error in recording supervisor for {device_id}: {e}")
 
     async def start_video_recording(
         self,
@@ -561,6 +565,7 @@ class UnifiedMobileController:
             local_video_path = output_dir / "recording.mkv"
             video_id = uuid4()
             start_time = time.time()
+            display_state = await get_android_display_state(device_id)
             data_engine_start_time = (
                 self.ctx.data_engine.session_start_time
                 if (self.ctx and self.ctx.data_engine)
@@ -573,8 +578,14 @@ class UnifiedMobileController:
                 start_time=start_time,
                 data_engine_start_time=data_engine_start_time,
                 local_video_path=local_video_path,
+                capture_width=getattr(getattr(self.ctx, "device", None), "device_width", None),
+                capture_height=getattr(getattr(self.ctx, "device", None), "device_height", None),
+                android_rotation=display_state[0] if display_state else None,
+                android_segment_started_at=start_time,
                 is_active=True,
             )
+            if display_state:
+                session.capture_width, session.capture_height = display_state[1:]
 
             # Persist to local database if Data Engine is active
             if self.ctx and self.ctx.data_engine:
@@ -586,24 +597,9 @@ class UnifiedMobileController:
                 )
 
             # Start scrcpy in background
-            cmd = [
-                "scrcpy",
-                "--serial",
-                device_id,
-                "--no-window",
-                "--record",
-                str(local_video_path),
-                "--record-format",
-                "mkv",
-                "--video-bit-rate",
-                "2M",
-            ]
+            cmd = build_scrcpy_record_command("scrcpy", device_id, local_video_path)
 
-            process = await asyncio.create_subprocess_exec(
-                *cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
+            process = await self._spawn_scrcpy(cmd)
 
             session.process = process
             set_active_session(device_id, session)
@@ -667,13 +663,13 @@ class UnifiedMobileController:
         session.is_active = False
         if session.watchdog_task and not session.watchdog_task.done():
             session.watchdog_task.cancel()
+            await session.watchdog_task
 
         try:
             process = session.process
             if process is not None:
                 try:
-                    process.terminate()
-                    await asyncio.wait_for(process.wait(), timeout=5.0)
+                    await self._stop_scrcpy(process)
                 except Exception as proc_e:
                     logger.warning(f"Error terminating scrcpy process: {proc_e}")
 
@@ -685,27 +681,26 @@ class UnifiedMobileController:
                     message="Recording file not found on disk",
                 )
 
-            # Concatenate previous segments if scrcpy crashed and restarted
-            if session.android_video_segments:
-                all_segments = session.android_video_segments
-                if output_path.exists() and output_path not in all_segments:
-                    all_segments.append(output_path)
-                concatenated_mkv = output_path.parent / "full_recording.mkv"
-                if await concatenate_videos(all_segments, concatenated_mkv):
-                    cleanup_video_segments(all_segments)
-                    output_path = concatenated_mkv
-
-            # Convert MKV to web-streamable MP4
-            mp4_path = (
-                output_path.parent / "recording.mp4"
-                if output_path.name == "full_recording.mkv"
-                else output_path.with_suffix(".mp4")
-            )
-            success = await self._convert_mkv_to_mp4(output_path, mp4_path)
+            self._finalize_current_segment(session, time.time())
+            for record in session.android_segment_records:
+                if not record.get("conversion_scheduled"):
+                    record["conversion_scheduled"] = True
+                    session.android_conversion_tasks.append(
+                        asyncio.create_task(
+                            self._remux_segment_record(record)
+                        )
+                    )
+            if session.android_conversion_tasks:
+                await asyncio.gather(*session.android_conversion_tasks, return_exceptions=True)
+            mp4_paths = [
+                Path(record["output_path"])
+                for record in session.android_segment_records
+                if Path(record["output_path"]).exists()
+            ]
+            await write_recording_manifest(output_path.parent, mp4_paths)
 
             remove_active_session(device_id)
-
-            final_video_path = mp4_path if (success and mp4_path.exists()) else output_path
+            final_video_path = mp4_paths[0] if mp4_paths else output_path
 
             # Persist update to local database if Data Engine is active
             if self.ctx and self.ctx.data_engine:
@@ -717,16 +712,11 @@ class UnifiedMobileController:
                     end_time=time.time(),
                 )
 
-            if success and mp4_path.exists():
-                try:
-                    if output_path != mp4_path and output_path.exists():
-                        output_path.unlink()
-                except Exception:
-                    pass
+            if mp4_paths:
                 return VideoRecordingResult(
                     success=True,
-                    message=f"Recording stopped, saved to {mp4_path}",
-                    video_path=mp4_path,
+                    message=f"Recording stopped, saved {len(mp4_paths)} orientation-safe segments",
+                    video_path=mp4_paths[0],
                 )
             else:
                 return VideoRecordingResult(
@@ -744,52 +734,19 @@ class UnifiedMobileController:
             )
 
     async def _convert_mkv_to_mp4(self, mkv_path: Path, mp4_path: Path) -> bool:
-        """Convert MKV to MP4 using ffmpeg with faststart for web streaming."""
-        if not mkv_path.exists():
-            return False
-        try:
-            # 1. Try fast stream copy with +faststart
-            process = await asyncio.create_subprocess_exec(
-                get_ffmpeg_path(),
-                "-y",
-                "-i",
-                str(mkv_path),
-                "-c",
-                "copy",
-                "-movflags",
-                "+faststart",
-                str(mp4_path),
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-            await process.wait()
-            if mp4_path.exists() and mp4_path.stat().st_size > 0:
-                return True
+        """Normalize MKV into a fixed-size, browser-safe MP4.
 
-            # 2. Fallback to re-encoding if stream copy fails
-            logger.warning("ffmpeg fast copy failed, re-encoding MKV to MP4...")
-            process = await asyncio.create_subprocess_exec(
-                get_ffmpeg_path(),
-                "-y",
-                "-i",
-                str(mkv_path),
-                "-c:v",
-                "libx264",
-                "-preset",
-                "fast",
-                "-pix_fmt",
-                "yuv420p",
-                "-movflags",
-                "+faststart",
-                str(mp4_path),
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-            await process.wait()
-            return mp4_path.exists() and mp4_path.stat().st_size > 0
-        except Exception as e:
-            logger.error(f"Failed to convert MKV to MP4: {e}")
-            return False
+        Stream-copying a scrcpy H.264 track is unsafe because historical or
+        recovered recordings may contain resolution changes. Re-encoding onto
+        the initial capture canvas guarantees one coded size for the full MP4.
+        """
+        session = get_active_session(self._get_device_id())
+        return await normalize_recording_to_mp4(
+            mkv_path,
+            mp4_path,
+            session.capture_width if session else None,
+            session.capture_height if session else None,
+        )
 
     async def cleanup(self) -> None:
         await self._driver.disconnect()
diff --git a/artemis/drivers/android/adb_driver.py b/artemis/drivers/android/adb_driver.py
index 5b11dd0..75e65a9 100644
--- a/artemis/drivers/android/adb_driver.py
+++ b/artemis/drivers/android/adb_driver.py
@@ -28,6 +28,7 @@ from artemis.clients.ui_automator_client import (
 from artemis.config.paths import get_temp_dir
 from artemis.drivers.base import BaseDeviceDriver, KeyCode, ScreenData, SwipeDirection
 from artemis.toolchain import find_ffmpeg, find_scrcpy
+from artemis.utils.video import build_scrcpy_record_command
 from artemis.utils.ui_filter import filter_ui_hierarchy
 from artemis.utils.logger import get_logger
 
@@ -114,6 +115,8 @@ class AndroidAdbDriver(BaseDeviceDriver):
         """Clean up ongoing recordings and tunnels."""
         if self._recording_process:
             await self.stop_video_recording()
+        if self._ui_adb_client and hasattr(self._ui_adb_client, "disconnect"):
+            await asyncio.to_thread(self._ui_adb_client.disconnect)
 
     async def get_screen_data(self, skip_settling: bool = False) -> ScreenData:
         """Captures screenshot and XML hierarchy concurrently."""
@@ -442,18 +445,12 @@ class AndroidAdbDriver(BaseDeviceDriver):
         logger.info(f"Starting scrcpy video recording to {self._recording_mkv_path}...")
         try:
             scrcpy_bin = find_scrcpy()
-            cmd = [
+            cmd = build_scrcpy_record_command(
                 scrcpy_bin,
-                "--serial",
                 self.device_id,
-                "--no-window",
-                "--record",
-                str(self._recording_mkv_path),
-                "--record-format",
-                "mkv",
-                "--video-bit-rate",
-                "2M",
-            ]
+                self._recording_mkv_path,
+                lock_capture_orientation=False,
+            )
             self._scrcpy_process = await asyncio.create_subprocess_exec(
                 *cmd,
                 stdout=asyncio.subprocess.PIPE,
diff --git a/artemis/utils/video.py b/artemis/utils/video.py
index 17b85e2..c31673f 100644
--- a/artemis/utils/video.py
+++ b/artemis/utils/video.py
@@ -19,8 +19,10 @@ Provides shared types and utilities for video recording across platforms.
 
 import asyncio
 import importlib.util
+import json
 from pathlib import Path
 import platform
+import re
 import shutil
 import subprocess
 from typing import Any
@@ -34,6 +36,7 @@ from artemis.utils.logger import get_logger
 logger = get_logger(__name__)
 
 DEFAULT_MAX_DURATION_SECONDS = 900  # 15 minutes
+ANDROID_RECORDING_SEGMENT_SECONDS = 1800
 VIDEO_READY_DELAY_SECONDS = 1
 ANDROID_DEVICE_VIDEO_PATH = "/sdcard/screen_recording.mp4"
 ANDROID_MAX_RECORDING_DURATION_SECONDS = 180  # Android screenrecord limit
@@ -44,6 +47,37 @@ MAX_VIDEO_SIZE_MB = 500
 MAX_VIDEO_SIZE_BYTES = MAX_VIDEO_SIZE_MB * 1024 * 1024
 
 
+def build_scrcpy_record_command(
+    scrcpy_executable: str,
+    device_id: str,
+    output_path: Path,
+    video_bit_rate: str = "2M",
+    lock_capture_orientation: bool = True,
+) -> list[str]:
+    """Build the shared scrcpy command used by all recording paths.
+
+    Each recording segment locks the orientation present when scrcpy starts.
+    The recording supervisor starts a new segment when Android rotates, so a
+    segment never contains multiple coded sizes while still displaying the app
+    in its natural orientation.
+    """
+    command = [
+        scrcpy_executable,
+        "--serial",
+        device_id,
+        "--no-window",
+        "--record",
+        str(output_path),
+        "--record-format",
+        "mkv",
+        "--video-bit-rate",
+        video_bit_rate,
+    ]
+    if lock_capture_orientation:
+        command.append("--capture-orientation=@")
+    return command
+
+
 class _CyFunctionDetectorMeta(type):
     def __instancecheck__(self, instance):
         name = type(instance).__name__
@@ -73,11 +107,17 @@ class RecordingSession(BaseModel):
     process: Any = None
     data_engine_start_time: float | None = None
     local_video_path: Path | None = None
+    capture_width: int | None = None
+    capture_height: int | None = None
     android_device_path: str = ANDROID_DEVICE_VIDEO_PATH
     android_video_segments: list[Path] = []
     android_segment_index: int = 0
     android_restart_task: asyncio.Task | None = None
     watchdog_task: asyncio.Task | None = None
+    android_rotation: int | None = None
+    android_segment_started_at: float | None = None
+    android_segment_records: list[dict[str, Any]] = []
+    android_conversion_tasks: list[asyncio.Task] = []
     is_active: bool = True
     errors: list[str] = []
 
@@ -138,6 +178,297 @@ def get_ffmpeg_path() -> str:
         return "ffmpeg"
 
 
+def get_ffprobe_path() -> str:
+    """Get ffprobe for lightweight segment metadata inspection."""
+    return shutil.which("ffprobe") or "ffprobe"
+
+
+async def normalize_recording_to_mp4(
+    source_path: Path,
+    output_path: Path,
+    capture_width: int | None,
+    capture_height: int | None,
+) -> bool:
+    """Re-encode a recording onto one fixed canvas for browser playback."""
+    if not source_path.exists():
+        return False
+
+    width = max(2, int(capture_width or 1080)) // 2 * 2
+    height = max(2, int(capture_height or 1920)) // 2 * 2
+    video_filter = (
+        f"scale={width}:{height}:force_original_aspect_ratio=decrease:"
+        f"force_divisible_by=2,pad={width}:{height}:"
+        "(ow-iw)/2:(oh-ih)/2:color=black,setsar=1"
+    )
+
+    try:
+        process = await asyncio.create_subprocess_exec(
+            get_ffmpeg_path(),
+            "-y",
+            "-fflags",
+            "+genpts",
+            "-i",
+            str(source_path),
+            "-map",
+            "0:v:0",
+            "-map",
+            "0:a?",
+            "-vf",
+            video_filter,
+            "-r",
+            "30",
+            "-fps_mode",
+            "cfr",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-crf",
+            "23",
+            "-pix_fmt",
+            "yuv420p",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        _stdout, stderr = await process.communicate()
+        if process.returncode == 0 and output_path.exists() and output_path.stat().st_size > 0:
+            return True
+
+        error_detail = stderr.decode(errors="replace")[-4000:]
+        logger.error(
+            f"Failed to normalize recording to MP4 (code {process.returncode}): {error_detail}"
+        )
+        if output_path.exists():
+            output_path.unlink()
+        return False
+    except Exception as e:
+        logger.error(f"Failed to normalize recording to MP4: {e}")
+        if output_path.exists():
+            output_path.unlink()
+        return False
+
+
+async def remux_recording_to_mp4(source_path: Path, output_path: Path) -> bool:
+    """Move a fixed-dimension recording segment into MP4 without re-encoding."""
+    if not source_path.exists() or source_path.stat().st_size == 0:
+        return False
+    try:
+        process = await asyncio.create_subprocess_exec(
+            get_ffmpeg_path(),
+            "-y",
+            "-fflags",
+            "+genpts",
+            "-i",
+            str(source_path),
+            "-map",
+            "0:v:0",
+            "-map",
+            "0:a?",
+            "-c",
+            "copy",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        _stdout, stderr = await process.communicate()
+        valid = process.returncode == 0 and output_path.exists() and output_path.stat().st_size > 0
+        if valid:
+            return True
+        logger.error(
+            f"Failed to remux recording segment (code {process.returncode}): "
+            f"{stderr.decode(errors='replace')[-2000:]}"
+        )
+    except Exception as e:
+        logger.error(f"Failed to remux recording segment: {e}")
+    if output_path.exists():
+        output_path.unlink()
+    return False
+
+
+async def probe_video_segment(video_path: Path) -> dict[str, float | int]:
+    """Read duration and coded dimensions for a finalized segment."""
+    process = await asyncio.create_subprocess_exec(
+        get_ffprobe_path(),
+        "-v",
+        "error",
+        "-select_streams",
+        "v:0",
+        "-show_entries",
+        "stream=width,height:format=duration",
+        "-of",
+        "json",
+        str(video_path),
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, _stderr = await process.communicate()
+    if process.returncode != 0:
+        return {}
+    try:
+        payload = json.loads(stdout)
+        stream = (payload.get("streams") or [{}])[0]
+        return {
+            "duration": float((payload.get("format") or {}).get("duration") or 0),
+            "width": int(stream.get("width") or 0),
+            "height": int(stream.get("height") or 0),
+        }
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return {}
+
+
+async def get_android_display_state(device_id: str) -> tuple[int, int, int] | None:
+    """Return Android's current (rotation, width, height)."""
+    try:
+        process = await asyncio.create_subprocess_exec(
+            "adb",
+            "-s",
+            device_id,
+            "shell",
+            "dumpsys",
+            "window",
+            "displays",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        stdout, _stderr = await asyncio.wait_for(process.communicate(), timeout=3.0)
+        text = stdout.decode(errors="replace")
+        rotation_match = re.search(r"\bmRotation=(\d+)", text)
+        size_match = re.search(r"\bcur=(\d+)x(\d+)", text)
+        if rotation_match and size_match:
+            return (
+                int(rotation_match.group(1)),
+                int(size_match.group(1)),
+                int(size_match.group(2)),
+            )
+    except (OSError, TimeoutError):
+        pass
+    return None
+
+
+async def write_recording_manifest(output_dir: Path, mp4_paths: list[Path]) -> Path | None:
+    """Write the browser playlist used for orientation-aware playback."""
+    segments = []
+    timeline = 0.0
+    for path in mp4_paths:
+        if not path.exists():
+            continue
+        metadata = await probe_video_segment(path)
+        duration = float(metadata.get("duration", 0))
+        segments.append(
+            {
+                "file": path.name,
+                "start": round(timeline, 3),
+                "duration": round(duration, 3),
+                "width": int(metadata.get("width", 0)),
+                "height": int(metadata.get("height", 0)),
+            }
+        )
+        timeline += duration
+    if not segments:
+        return None
+    manifest_path = output_dir / "recording.json"
+    manifest_path.write_text(
+        json.dumps(
+            {"version": 1, "duration": round(timeline, 3), "segments": segments},
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+    return manifest_path
+
+
+async def render_timeline_clip(
+    segments: list[dict[str, Any]],
+    start_time: float,
+    end_time: float,
+    output_path: Path,
+    canvas_width: int = 720,
+    canvas_height: int = 1280,
+) -> bool:
+    """Render one continuous analyzer clip from orientation-aware segments.
+
+    Only the requested time window is decoded. This keeps the video analyzer's
+    single-MP4 contract without ever re-encoding the full recording.
+    """
+    overlaps = []
+    for segment in segments:
+        path = Path(segment["path"])
+        seg_start = float(segment["start"])
+        seg_end = float(segment["end"])
+        overlap_start = max(start_time, seg_start)
+        overlap_end = min(end_time, seg_end)
+        if path.exists() and overlap_end > overlap_start:
+            overlaps.append((path, overlap_start - seg_start, overlap_end - seg_start))
+    if not overlaps:
+        return False
+
+    command = [get_ffmpeg_path(), "-y", "-fflags", "+genpts"]
+    for path, _start, _end in overlaps:
+        command.extend(["-i", str(path)])
+
+    filter_parts = []
+    labels = []
+    for index, (_path, local_start, local_end) in enumerate(overlaps):
+        label = f"v{index}"
+        filter_parts.append(
+            f"[{index}:v:0]trim=start={local_start:.6f}:end={local_end:.6f},"
+            "setpts=PTS-STARTPTS,"
+            f"scale={canvas_width}:{canvas_height}:force_original_aspect_ratio=decrease:"
+            "force_divisible_by=2,"
+            f"pad={canvas_width}:{canvas_height}:(ow-iw)/2:(oh-ih)/2:color=black,"
+            f"setsar=1,fps=15,format=yuv420p[{label}]"
+        )
+        labels.append(f"[{label}]")
+    if len(labels) == 1:
+        filter_parts.append(f"{labels[0]}null[outv]")
+    else:
+        filter_parts.append(f"{''.join(labels)}concat=n={len(labels)}:v=1:a=0[outv]")
+
+    command.extend(
+        [
+            "-filter_complex",
+            ";".join(filter_parts),
+            "-map",
+            "[outv]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-crf",
+            "23",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+        ]
+    )
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        _stdout, stderr = await process.communicate()
+        if process.returncode == 0 and output_path.exists() and output_path.stat().st_size > 0:
+            return True
+        logger.error(
+            f"Failed to render analyzer timeline clip (code {process.returncode}): "
+            f"{stderr.decode(errors='replace')[-3000:]}"
+        )
+    except Exception as e:
+        logger.error(f"Failed to render analyzer timeline clip: {e}")
+    if output_path.exists():
+        output_path.unlink()
+    return False
+
 def is_scrcpy_installed() -> bool:
     """Check if scrcpy is available in the system PATH."""
 
diff --git a/tests/unit/test_ui_automator_client_awake.py b/tests/unit/test_ui_automator_client_awake.py
new file mode 100644
index 0000000..fd9ed57
--- /dev/null
+++ b/tests/unit/test_ui_automator_client_awake.py
@@ -0,0 +1,165 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from artemis.clients.ui_automator_client import (
+    AWAKE_STRATEGY_HEARTBEAT,
+    AWAKE_STRATEGY_STAY_ON,
+    AWAKE_STRATEGY_WAKE_LOCK,
+    UIAutomatorClient,
+    _keep_device_awake,
+)
+from artemis.drivers.android.adb_driver import AndroidAdbDriver
+
+
+def completed(stdout="", returncode=0, stderr=""):
+    return MagicMock(returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+@patch("artemis.clients.ui_automator_client.subprocess.run")
+def test_keep_device_awake_uses_verified_primary_strategy(mock_run):
+    mock_run.side_effect = [
+        completed(),
+        completed(),
+        completed(),
+        completed("  mStayOn=true\n"),
+    ]
+
+    strategy = _keep_device_awake("device-123")
+
+    assert strategy == AWAKE_STRATEGY_STAY_ON
+    assert mock_run.call_count == 4
+
+
+@patch("artemis.clients.ui_automator_client.subprocess.run")
+def test_keep_device_awake_falls_back_to_verified_screen_wake_lock(mock_run):
+    mock_run.side_effect = [
+        completed(),
+        completed(),
+        completed(),
+        completed("  mStayOn=false\n"),
+        completed("WakeLock{abc held=true, refCount=1}\n"),
+        completed(
+            "Display 0, wakelock type: SCREEN_BRIGHT_WAKE_LOCK: "
+            "WakeLock{abc held=true, refCount=1}\n"
+        ),
+    ]
+
+    strategy = _keep_device_awake("device-123")
+
+    assert strategy == AWAKE_STRATEGY_WAKE_LOCK
+    assert mock_run.call_args_list[-2].args[0] == [
+        "adb",
+        "-s",
+        "device-123",
+        "shell",
+        "cmd",
+        "power",
+        "set-wakelock",
+        "acquire",
+        "-d",
+        "0",
+        "SCREEN_BRIGHT_WAKE_LOCK",
+    ]
+
+
+@patch("artemis.clients.ui_automator_client.subprocess.run")
+def test_keep_device_awake_uses_heartbeat_when_wake_lock_is_unavailable(mock_run):
+    mock_run.side_effect = [
+        completed(),
+        completed(),
+        completed(),
+        completed("  mStayOn=false\n"),
+        completed(returncode=1, stderr="unknown command"),
+        completed("Wakelocks:\n"),
+    ]
+
+    strategy = _keep_device_awake("device-123")
+
+    assert strategy == AWAKE_STRATEGY_HEARTBEAT
+
+
+@patch.dict("os.environ", {"ARTEMIS_KEEP_DEVICE_AWAKE": "false"})
+@patch("artemis.clients.ui_automator_client.subprocess.run")
+def test_keep_device_awake_can_be_disabled(mock_run):
+    assert _keep_device_awake("device-123") is None
+    mock_run.assert_not_called()
+
+
+@patch("artemis.clients.ui_automator_client.u2.connect")
+@patch(
+    "artemis.clients.ui_automator_client._keep_device_awake",
+    return_value=AWAKE_STRATEGY_HEARTBEAT,
+)
+@patch("artemis.clients.ui_automator_client._ensure_maestro_not_installed")
+def test_new_ui_connection_starts_heartbeat_fallback(
+    mock_remove_maestro, mock_keep_awake, mock_connect
+):
+    client = UIAutomatorClient("device-123")
+    client._start_awake_heartbeat = MagicMock()
+
+    client._ensure_connected()
+
+    mock_remove_maestro.assert_called_once_with("device-123")
+    mock_keep_awake.assert_called_once_with("device-123")
+    mock_connect.assert_called_once_with("device-123")
+    client._start_awake_heartbeat.assert_called_once_with()
+    assert client._awake_strategy == AWAKE_STRATEGY_HEARTBEAT
+
+
+@patch("artemis.clients.ui_automator_client._release_screen_wake_lock")
+def test_disconnect_releases_screen_wake_lock(mock_release):
+    client = UIAutomatorClient("device-123")
+    client._awake_strategy = AWAKE_STRATEGY_WAKE_LOCK
+    client._stop_awake_heartbeat = MagicMock()
+
+    client.disconnect()
+
+    client._stop_awake_heartbeat.assert_called_once_with()
+    mock_release.assert_called_once_with("device-123")
+    assert client._awake_strategy is None
+
+
+@patch("artemis.clients.ui_automator_client.threading.Thread")
+@patch("artemis.clients.ui_automator_client.threading.Event")
+def test_heartbeat_thread_is_daemonized_and_stoppable(mock_event, mock_thread):
+    stop_event = MagicMock()
+    thread = MagicMock()
+    mock_event.return_value = stop_event
+    mock_thread.return_value = thread
+    client = UIAutomatorClient("device-123")
+
+    client._start_awake_heartbeat()
+    client._stop_awake_heartbeat()
+
+    assert mock_thread.call_args.kwargs["daemon"] is True
+    thread.start.assert_called_once_with()
+    stop_event.set.assert_called_once_with()
+    thread.join.assert_called_once_with(timeout=2.0)
+
+
+@pytest.mark.asyncio
+async def test_android_driver_disconnect_cleans_up_ui_client():
+    ui_client = MagicMock()
+    driver = AndroidAdbDriver(
+        device_id="device-123",
+        adb_client=MagicMock(),
+        ui_adb_client=ui_client,
+    )
+
+    await driver.disconnect()
+
+    ui_client.disconnect.assert_called_once_with()
+
+
+@patch("artemis.clients.ui_automator_client.subprocess.run")
+def test_keep_device_awake_continues_after_an_adb_failure(mock_run):
+    mock_run.side_effect = [
+        completed(returncode=1, stderr="not allowed"),
+        OSError("adb temporarily unavailable"),
+        completed(),
+        completed("mStayOn=true\n"),
+    ]
+
+    assert _keep_device_awake("device-123") == AWAKE_STRATEGY_STAY_ON
+    assert mock_run.call_count == 4
diff --git a/tests/unit/test_unified_controller_video.py b/tests/unit/test_unified_controller_video.py
index c344f75..9e8d0d4 100644
--- a/tests/unit/test_unified_controller_video.py
+++ b/tests/unit/test_unified_controller_video.py
@@ -14,23 +14,161 @@
 
 """Unit tests for UnifiedMobileController video recording and playback features."""
 
+import subprocess
 import time
 from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
 import pytest
+import cv2
 
 from artemis.context import ArtemisContext
 from artemis.controllers.unified_controller import UnifiedMobileController
 from artemis.drivers.mock.mock_driver import MockDeviceDriver
 from artemis.utils.video import (
     RecordingSession,
+    build_scrcpy_record_command,
+    get_ffmpeg_path,
     get_active_session,
+    normalize_recording_to_mp4,
+    render_timeline_clip,
     remove_active_session,
     set_active_session,
 )
 
 
+def test_scrcpy_recording_locks_each_segment_orientation(tmp_path):
+    output_path = tmp_path / "recording.mkv"
+
+    command = build_scrcpy_record_command("scrcpy", "device-1", output_path)
+
+    assert "--capture-orientation=@" in command
+    assert command[command.index("--record") + 1] == str(output_path)
+
+
+@pytest.mark.asyncio
+async def test_normalize_recording_handles_resolution_change(tmp_path):
+    """A portrait/landscape H.264 track must become one fixed-size MP4."""
+    ffmpeg = get_ffmpeg_path()
+    portrait = tmp_path / "portrait.mkv"
+    landscape = tmp_path / "landscape.mkv"
+    dynamic = tmp_path / "dynamic.mkv"
+    output = tmp_path / "recording.mp4"
+
+    for size, path in (("108x242", portrait), ("242x108", landscape)):
+        subprocess.run(
+            [
+                ffmpeg,
+                "-y",
+                "-f",
+                "lavfi",
+                "-i",
+                f"testsrc=size={size}:rate=10:duration=0.5",
+                "-c:v",
+                "libx264",
+                "-pix_fmt",
+                "yuv420p",
+                str(path),
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+    concat_file = tmp_path / "segments.txt"
+    concat_file.write_text(
+        f"file '{portrait.as_posix()}'\nfile '{landscape.as_posix()}'\n",
+        encoding="utf-8",
+    )
+    subprocess.run(
+        [
+            ffmpeg,
+            "-y",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            str(concat_file),
+            "-c",
+            "copy",
+            str(dynamic),
+        ],
+        check=True,
+        capture_output=True,
+    )
+
+    assert await normalize_recording_to_mp4(dynamic, output, 108, 242)
+
+    capture = cv2.VideoCapture(str(output))
+    dimensions = set()
+    frame_count = 0
+    while True:
+        ok, frame = capture.read()
+        if not ok:
+            break
+        frame_count += 1
+        dimensions.add((frame.shape[1], frame.shape[0]))
+    capture.release()
+
+    assert frame_count >= 8
+    assert dimensions == {(108, 242)}
+
+
+@pytest.mark.asyncio
+async def test_analyzer_clip_stays_continuous_across_orientation_segments(tmp_path):
+    """The analyzer must still receive one decodable MP4 across a rotation."""
+    ffmpeg = get_ffmpeg_path()
+    portrait = tmp_path / "recording.mkv"
+    landscape = tmp_path / "recording_001.mkv"
+    output = tmp_path / "agent_clip.mp4"
+    for color, size, path in (
+        ("red", "108x242", portrait),
+        ("blue", "242x108", landscape),
+    ):
+        subprocess.run(
+            [
+                ffmpeg,
+                "-y",
+                "-f",
+                "lavfi",
+                "-i",
+                f"color={color}:size={size}:rate=15:duration=1",
+                "-c:v",
+                "libx264",
+                "-pix_fmt",
+                "yuv420p",
+                str(path),
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+    assert await render_timeline_clip(
+        [
+            {"path": portrait, "start": 0.0, "end": 1.0},
+            {"path": landscape, "start": 1.0, "end": 2.0},
+        ],
+        0.25,
+        1.75,
+        output,
+        canvas_width=180,
+        canvas_height=320,
+    )
+
+    capture = cv2.VideoCapture(str(output))
+    dimensions = set()
+    frames = 0
+    while True:
+        ok, frame = capture.read()
+        if not ok:
+            break
+        dimensions.add((frame.shape[1], frame.shape[0]))
+        frames += 1
+    capture.release()
+    assert dimensions == {(180, 320)}
+    assert frames >= 20
+
+
 @pytest.fixture
 def mock_ctx(tmp_path):
     ctx = MagicMock(spec=ArtemisContext)
@@ -60,17 +198,21 @@ async def test_unified_controller_start_recording(mock_ctx, tmp_path):
     mock_proc.returncode = None
 
     with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=mock_proc)):
-        with patch("asyncio.sleep", AsyncMock()):
-            res = await controller.start_video_recording(output_dir=tmp_path)
+        with patch(
+            "artemis.controllers.unified_controller.get_android_display_state",
+            AsyncMock(return_value=(0, 1080, 2424)),
+        ):
+            with patch("asyncio.sleep", AsyncMock()):
+                res = await controller.start_video_recording(output_dir=tmp_path)
 
-            assert res.success is True
-            assert get_active_session("emulator-5554") is not None
-            assert mock_ctx.data_engine.record_video_start.called
+                assert res.success is True
+                assert get_active_session("emulator-5554") is not None
+                assert mock_ctx.data_engine.record_video_start.called
 
-            # Calling start again should report already in progress
-            res2 = await controller.start_video_recording(output_dir=tmp_path)
-            assert res2.success is False
-            assert "already in progress" in res2.message
+                # Calling start again should report already in progress
+                res2 = await controller.start_video_recording(output_dir=tmp_path)
+                assert res2.success is False
+                assert "already in progress" in res2.message
 
     remove_active_session("emulator-5554")
 
@@ -99,10 +241,11 @@ async def test_unified_controller_stop_recording(mock_ctx, tmp_path):
     )
     set_active_session("emulator-5554", session)
 
-    with patch.object(
-        controller,
-        "_convert_mkv_to_mp4",
+    with patch(
+        "artemis.controllers.unified_controller.remux_recording_to_mp4",
         AsyncMock(side_effect=lambda src, dst: dst.write_bytes(b"mp4 content") or True),
+    ), patch(
+        "artemis.controllers.unified_controller.write_recording_manifest", AsyncMock()
     ):
         res = await controller.stop_video_recording()
 
@@ -135,7 +278,7 @@ async def test_unified_controller_extract_segment_metadata(mock_ctx, tmp_path):
     set_active_session("emulator-5554", session)
 
     with patch(
-        "artemis.controllers.unified_controller.trim_video",
+        "artemis.controllers.unified_controller.render_timeline_clip",
         AsyncMock(side_effect=lambda src, s, e, dst: dst.write_bytes(b"segment mp4") or True),
     ):
         res = await controller.extract_segment_metadata(start_time=2.0, end_time=8.0)
@@ -204,6 +347,14 @@ async def test_unified_controller_crash_recovery_and_multi_segment(mock_ctx, tmp
         data_engine_start_time=time.time() - 30.0,
         local_video_path=seg2,
         android_video_segments=[seg1],
+        android_segment_records=[
+            {
+                "path": seg1,
+                "output_path": tmp_path / "recording.mp4",
+                "start": 0.0,
+                "end": 15.0,
+            }
+        ],
         android_segment_index=1,
         process=mock_proc,
     )
@@ -211,14 +362,10 @@ async def test_unified_controller_crash_recovery_and_multi_segment(mock_ctx, tmp
 
     with (
         patch(
-            "artemis.controllers.unified_controller.concatenate_videos",
-            AsyncMock(side_effect=lambda segs, out: out.write_bytes(b"combined mkv") or True),
-        ),
-        patch.object(
-            controller,
-            "_convert_mkv_to_mp4",
+            "artemis.controllers.unified_controller.remux_recording_to_mp4",
             AsyncMock(side_effect=lambda src, dst: dst.write_bytes(b"final mp4") or True),
         ),
+        patch("artemis.controllers.unified_controller.write_recording_manifest", AsyncMock()),
     ):
         res = await controller.stop_video_recording()
 
@@ -261,7 +408,8 @@ async def test_unified_controller_timeline_alignment(mock_ctx, tmp_path):
         return True
 
     with patch(
-        "artemis.controllers.unified_controller.trim_video", AsyncMock(side_effect=mock_trim)
+        "artemis.controllers.unified_controller.render_timeline_clip",
+        AsyncMock(side_effect=mock_trim),
     ):
         # Agent asks for system time range [5.0s, 10.0s]
         res = await controller.extract_segment_metadata(start_time=5.0, end_time=10.0)