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)