diff --git a/.github/workflows/recording-reliability.yml b/.github/workflows/recording-reliability.yml
index 9196e58fb9..9b2ddfc6c3 100644
--- a/.github/workflows/recording-reliability.yml
+++ b/.github/workflows/recording-reliability.yml
@@ -47,4 +47,6 @@ jobs:
             __tests__/unit/media-server-progress.test.ts \
             __tests__/unit/media-processing-budget.test.ts \
             __tests__/unit/playback-source.test.ts \
-            __tests__/unit/upload-progress-playback.test.ts
+            __tests__/unit/upload-progress-playback.test.ts \
+            __tests__/unit/recording-prepare.test.ts \
+            __tests__/unit/s3-bucket-connections.test.ts
diff --git a/apps/desktop-gpui/src/upload.rs b/apps/desktop-gpui/src/upload.rs
index d84d09ddf7..f186f10c99 100644
--- a/apps/desktop-gpui/src/upload.rs
+++ b/apps/desktop-gpui/src/upload.rs
@@ -7,6 +7,7 @@ use std::time::{Duration, Instant};
 
 use cap_enc_ffmpeg::segmented_stream::{SegmentCompletedEvent, SegmentMediaType};
 use cap_project::{RecordingMeta, S3UploadMeta, SharingMeta, UploadMeta, VideoUploadInfo};
+use cap_recording::upload_preparation::{Preparation, Segment};
 use futures_util::{StreamExt as _, stream::FuturesUnordered};
 use reqwest::StatusCode;
 use serde::{Deserialize, Serialize};
@@ -517,6 +518,13 @@ async fn run_segment_upload(
 }
 
 trait SegmentTransport: Send + Sync {
+    fn prepare(
+        &self,
+        _video_id: &str,
+        _segments: Vec<Segment>,
+    ) -> impl Future<Output = Result<Option<Vec<Segment>>, String>> + Send {
+        std::future::ready(Ok(None))
+    }
     fn prefetch(
         &self,
         video_id: &str,
@@ -545,6 +553,41 @@ trait SegmentTransport: Send + Sync {
 
 struct LiveSegmentTransport;
 impl SegmentTransport for LiveSegmentTransport {
+    async fn prepare(
+        &self,
+        video_id: &str,
+        segments: Vec<Segment>,
+    ) -> Result<Option<Vec<Segment>>, String> {
+        tokio::time::timeout(Duration::from_secs(20), async {
+            #[derive(Deserialize)]
+            struct Response {
+                version: u32,
+                prepared: Vec<Segment>,
+            }
+            let response = auth::authed_request(
+                reqwest::Method::POST,
+                "/api/recording/prepare",
+                Some(json!({ "videoId": video_id, "segments": segments })),
+            )
+            .await
+            .map_err(|error| error.to_string())?;
+            let status = response.status();
+            if status == StatusCode::NOT_FOUND || status == StatusCode::METHOD_NOT_ALLOWED {
+                return Ok(None);
+            }
+            if !status.is_success() {
+                return Err(format!("Recording preparation returned {status}"));
+            }
+            let response = response
+                .json::<Response>()
+                .await
+                .map_err(|error| error.to_string())?;
+            Ok((response.version == 1 && response.prepared.len() <= 32)
+                .then_some(response.prepared))
+        })
+        .await
+        .map_err(|error| error.to_string())?
+    }
     async fn prefetch(
         &self,
         video_id: &str,
@@ -589,6 +632,13 @@ async fn upload_segments(
     let mut uploads = FuturesUnordered::new();
     let mut events_closed = false;
     let mut last_manifest_upload: Option<Instant> = None;
+    let mut preparation = Preparation::default();
+    let mut preparation_enabled = true;
+    let mut preparation_batch = Vec::new();
+    let mut preparation_interval = tokio::time::interval(Duration::from_secs(30));
+    preparation_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut preparation_request = None;
+    let mut next_preparation_request = tokio::time::Instant::now();
     let mut next_prefetch = SEGMENT_URL_PREFETCH + 1;
     let prefetched = checked_segment_step(&cancel, || async {
         Ok(transport.prefetch(video_id, 1, SEGMENT_URL_PREFETCH).await)
@@ -612,6 +662,25 @@ async fn upload_segments(
             None => (None, None),
         };
         tokio::select! {
+            _ = preparation_interval.tick(), if preparation_enabled && preparation_request.is_none() => {
+                if tokio::time::Instant::now() < next_preparation_request { continue; }
+                preparation_batch = preparation.next_batch(manifest.video_segments.iter().map(|segment| segment.index), manifest.audio_segments.iter().map(|segment| segment.index));
+                if !preparation_batch.is_empty() {
+                    preparation_request = Some(Box::pin(transport.prepare(video_id, preparation_batch.clone())));
+                }
+            }
+            response = async { preparation_request.as_mut().unwrap().await }, if preparation_request.is_some() => {
+                preparation_request = None;
+                match response {
+                    Ok(Some(prepared)) => preparation.acknowledge(&preparation_batch, &prepared),
+                    Ok(None) => preparation_enabled = false,
+                    Err(error) => {
+                        preparation.request_failed();
+                        tracing::debug!(%error, "Optional recording preparation unavailable");
+                    }
+                }
+                next_preparation_request = tokio::time::Instant::now() + preparation.retry_delay();
+            }
             permission = async { permission.unwrap().await }, if !authorized => {
                 permission.map_err(|_| "Instant completion was not authorized".to_string())?;
                 authorized = true;
@@ -667,6 +736,7 @@ async fn upload_segments(
         }
     }
 
+    drop(preparation_request);
     if cancel.load(Ordering::Acquire) {
         return Err("Instant recording upload cancelled".to_string());
     }
@@ -2354,6 +2424,9 @@ mod tests {
     }
     #[derive(Default)]
     struct FakeSegmentTransport {
+        delay_preparation: AtomicBool,
+        preparation_started: tokio::sync::Notify,
+        preparation_dropped: AtomicBool,
         manifests: Mutex<Vec<bool>>,
         completed: std::sync::atomic::AtomicUsize,
         uploaded: std::sync::atomic::AtomicUsize,
@@ -2367,6 +2440,20 @@ mod tests {
         prefetch_response: tokio::sync::Notify,
     }
     impl SegmentTransport for FakeSegmentTransport {
+        async fn prepare(&self, _: &str, _: Vec<Segment>) -> Result<Option<Vec<Segment>>, String> {
+            if !self.delay_preparation.load(Ordering::Acquire) {
+                return Ok(None);
+            }
+            struct Finish<'a>(&'a AtomicBool);
+            impl Drop for Finish<'_> {
+                fn drop(&mut self) {
+                    self.0.store(true, Ordering::Release);
+                }
+            }
+            let _finish = Finish(&self.preparation_dropped);
+            self.preparation_started.notify_one();
+            std::future::pending().await
+        }
         async fn prefetch(
             &self,
             _: &str,
@@ -2412,6 +2499,40 @@ mod tests {
             Ok(())
         }
     }
+    #[tokio::test]
+    async fn stopped_upload_drops_optional_preparation_without_waiting_for_it() {
+        let transport = FakeSegmentTransport::default();
+        transport.delay_preparation.store(true, Ordering::Release);
+        let (sender, events) = flume::unbounded();
+        sender
+            .send(segment_event(0, 0.0, true, SegmentMediaType::Video))
+            .unwrap();
+        sender
+            .send(segment_event(1, 1.0, false, SegmentMediaType::Video))
+            .unwrap();
+        let upload = upload_segments(
+            &transport,
+            "preparation",
+            events,
+            Arc::new(AtomicBool::new(false)),
+            None,
+        );
+        tokio::pin!(upload);
+        tokio::select! {
+            _ = transport.preparation_started.notified() => {}
+            result = &mut upload => panic!("Upload ended before preparation: {result:?}"),
+            _ = tokio::time::sleep(Duration::from_secs(35)) => panic!("Preparation did not start"),
+        }
+        drop(sender);
+        tokio::time::timeout(Duration::from_secs(1), upload)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(transport.preparation_dropped.load(Ordering::Acquire));
+        assert_eq!(transport.completed.load(Ordering::Acquire), 1);
+        assert_eq!(transport.manifests.lock().unwrap().last(), Some(&true));
+    }
+
     fn closed_segment_events() -> flume::Receiver<SegmentCompletedEvent> {
         let (sender, receiver) = flume::unbounded();
         sender
diff --git a/apps/desktop/src-tauri/src/api.rs b/apps/desktop/src-tauri/src/api.rs
index f0c598b4ab..1aa56dfd69 100644
--- a/apps/desktop/src-tauri/src/api.rs
+++ b/apps/desktop/src-tauri/src/api.rs
@@ -329,6 +329,42 @@ pub struct Organization {
     pub brand_colors: OrganizationBrandColors,
 }
 
+pub(crate) async fn prepare_recording_segments(
+    app: &AppHandle,
+    video_id: &str,
+    segments: &[crate::upload::preparation::Segment],
+) -> Result<Option<Vec<crate::upload::preparation::Segment>>, AuthedApiError> {
+    #[derive(Deserialize)]
+    struct Response {
+        version: u32,
+        prepared: Vec<crate::upload::preparation::Segment>,
+    }
+
+    let response = app
+        .authed_api_request("/api/recording/prepare", |client, url| {
+            client
+                .post(url)
+                .timeout(std::time::Duration::from_secs(20))
+                .json(&serde_json::json!({ "videoId": video_id, "segments": segments }))
+        })
+        .await?;
+    if matches!(response.status().as_u16(), 404 | 405) {
+        return Ok(None);
+    }
+    if !response.status().is_success() {
+        return Err(format!(
+            "Optional recording preparation unavailable ({})",
+            response.status()
+        )
+        .into());
+    }
+    let response: Response = crate::upload::lifecycle::cancellable(response.json()).await??;
+    if response.version != 1 || response.prepared.len() > 32 {
+        return Ok(None);
+    }
+    Ok(Some(response.prepared))
+}
+
 pub async fn verify_recording_complete(
     app: &AppHandle,
     video_id: &str,
diff --git a/apps/desktop/src-tauri/src/upload.rs b/apps/desktop/src-tauri/src/upload.rs
index 3ad9824207..757ed51cee 100644
--- a/apps/desktop/src-tauri/src/upload.rs
+++ b/apps/desktop/src-tauri/src/upload.rs
@@ -47,6 +47,7 @@ use tokio_util::io::ReaderStream;
 use tracing::{Span, debug, error, info, info_span, instrument, trace, warn};
 
 pub(crate) mod lifecycle;
+pub(crate) mod preparation;
 pub(crate) mod resume;
 use tracing_futures::Instrument;
 
@@ -1337,6 +1338,12 @@ impl SegmentUploader {
         })?;
 
         let state = Arc::new(Mutex::new(SegmentUploadState::new()));
+        let preparation = preparation::start(
+            app.clone(),
+            video_id.clone(),
+            state.clone(),
+            session.clone(),
+        );
         let semaphore = Arc::new(tokio::sync::Semaphore::new(6));
         let read_semaphore = Arc::new(tokio::sync::Semaphore::new(12));
         let consecutive_failures = Arc::new(std::sync::atomic::AtomicU32::new(0));
@@ -1666,6 +1673,7 @@ impl SegmentUploader {
         }
 
         drain_segment_upload_tasks(&state, &mut in_flight).await;
+        preparation.stop().await;
 
         if bridge_handle.join().is_err() {
             state
diff --git a/apps/desktop/src-tauri/src/upload/preparation.rs b/apps/desktop/src-tauri/src/upload/preparation.rs
new file mode 100644
index 0000000000..f8a5025a28
--- /dev/null
+++ b/apps/desktop/src-tauri/src/upload/preparation.rs
@@ -0,0 +1,96 @@
+use super::{SegmentUploadState, lifecycle};
+use crate::{api, web_api::inherit_upload_context};
+use cap_recording::upload_preparation::Preparation;
+pub(crate) use cap_recording::upload_preparation::Segment;
+use std::{
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+use tauri::AppHandle;
+
+pub(super) struct Task(tokio::task::JoinHandle<()>);
+
+impl Task {
+    pub(super) async fn stop(mut self) {
+        self.0.abort();
+        if let Err(error) = (&mut self.0).await
+            && !error.is_cancelled()
+        {
+            tracing::warn!(%error, "Optional recording preparation stopped");
+        }
+    }
+}
+
+impl Drop for Task {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
+pub(super) fn start(
+    app: AppHandle,
+    video_id: String,
+    state: Arc<Mutex<SegmentUploadState>>,
+    session: Arc<lifecycle::Session>,
+) -> Task {
+    let context = session.context();
+    Task(tokio::spawn(inherit_upload_context(context, async move {
+        let mut preparation = Preparation::default();
+        let mut next_request = tokio::time::Instant::now();
+        let mut interval = tokio::time::interval(Duration::from_secs(30));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        loop {
+            tokio::select! {
+                _ = session.cancelled() => return,
+                _ = interval.tick() => {}
+            }
+            if tokio::time::Instant::now() < next_request {
+                continue;
+            }
+            let batch = {
+                let state = state.lock().unwrap_or_else(|error| error.into_inner());
+                preparation.next_batch(
+                    state.uploaded_video_segments.keys().copied(),
+                    state.uploaded_audio_segments.keys().copied(),
+                )
+            };
+            if batch.is_empty() {
+                continue;
+            }
+            let result = tokio::select! {
+                _ = session.cancelled() => return,
+                result = api::prepare_recording_segments(&app, &video_id, &batch) => result,
+            };
+            match result {
+                Ok(Some(prepared)) => {
+                    preparation.acknowledge(&batch, &prepared);
+                }
+                Ok(None) => return,
+                Err(error) => {
+                    preparation.request_failed();
+                    tracing::debug!(%error, "Optional recording preparation unavailable");
+                }
+            }
+            next_request = tokio::time::Instant::now() + preparation.retry_delay();
+        }
+    })))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn stopping_preparation_cancels_an_in_flight_request() {
+        let (started, ready) = tokio::sync::oneshot::channel();
+        let handle = tokio::spawn(async move {
+            started.send(()).unwrap();
+            std::future::pending::<()>().await;
+        });
+        let abort = handle.abort_handle();
+        let task = Task(handle);
+        ready.await.unwrap();
+        task.stop().await;
+        assert!(abort.is_finished());
+    }
+}
diff --git a/apps/web/__tests__/integration/desktop-recording-production-replay.test.ts b/apps/web/__tests__/integration/desktop-recording-production-replay.test.ts
new file mode 100644
index 0000000000..4f810487ce
--- /dev/null
+++ b/apps/web/__tests__/integration/desktop-recording-production-replay.test.ts
@@ -0,0 +1,275 @@
+import { createHash } from "node:crypto";
+import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+import { Effect, Option } from "effect";
+import { describe, expect, it, vi } from "vitest";
+import { z } from "zod";
+
+const mocks = vi.hoisted(() => ({ storage: vi.fn() }));
+vi.mock("@cap/web-backend/src/Storage/index", () => ({
+	Storage: { getAccessForVideo: mocks.storage },
+}));
+vi.mock("@/lib/workflow-runtime", async () => {
+	const { Effect } = await import("effect");
+	return { runWorkflowPromise: Effect.runPromise };
+});
+vi.mock("@/lib/video-storage", () => ({
+	decodeStorageVideo: (video: unknown) => video,
+}));
+
+import {
+	buildDesktopRecordingSourceUrls,
+	commitDesktopRecordingSource,
+	prepareDesktopRecordingSegments,
+} from "@/lib/desktop-recording-source";
+import { readCompletedRecordingManifest } from "@/lib/desktop-recording-verification";
+
+const corpusRoot = process.env.CAP_RECORDING_REPLAY_CORPUS;
+const schema = z.array(
+	z.object({
+		label: z.string().regex(/^[a-zA-Z0-9_-]+$/),
+		provider: z.enum(["s3", "googleDrive"]),
+		manifest: z.record(z.unknown()),
+		fragments: z.array(
+			z.object({
+				track: z.enum(["video", "audio"]),
+				index: z.number().int().nonnegative(),
+				file: z.string().regex(/^[a-zA-Z0-9_-]+\/[a-zA-Z0-9_.-]+$/),
+				size: z.number().int().positive(),
+				sha256: z.string().regex(/^[a-f0-9]{64}$/),
+				identity: z.string(),
+			}),
+		),
+	}),
+);
+const corpus = corpusRoot
+	? schema.parse(
+			JSON.parse(readFileSync(join(corpusRoot, "index.json"), "utf8")),
+		)
+	: [];
+type Sample = (typeof corpus)[number];
+type Stored = {
+	body: Buffer;
+	identity: string;
+	metadata?: Record<string, string>;
+};
+const prefix = "replay-owner/replay-recording";
+const video = {
+	id: "replay-recording",
+	ownerId: "replay-owner",
+	source: { type: "desktopSegments" },
+} as Parameters<typeof commitDesktopRecordingSource>[0];
+const hash = (body: Buffer) => createHash("sha256").update(body).digest("hex");
+const key = (track: "video" | "audio", index: number) =>
+	`${prefix}/segments/${track}/${index === 0 ? "init.mp4" : `segment_${String(index).padStart(3, "0")}.m4s`}`;
+
+function replay(sample: Sample) {
+	if (!corpusRoot) throw new Error("Production corpus directory is required");
+	const objects = new Map<string, Stored>();
+	for (const fragment of sample.fragments) {
+		const body = readFileSync(join(corpusRoot, fragment.file));
+		expect(body.length).toBe(fragment.size);
+		expect(hash(body)).toBe(fragment.sha256);
+		objects.set(key(fragment.track, fragment.index), {
+			body,
+			identity: fragment.identity,
+		});
+	}
+	const object = (name: string) => {
+		const value = objects.get(name);
+		if (!value)
+			throw Object.assign(new Error("Source missing"), { name: "NoSuchKey" });
+		return value;
+	};
+	const checked = <T>(operation: () => T) => Effect.try(operation);
+	const bucket = {
+		provider: sample.provider,
+		bucketName: "replay",
+		headObject: (name: string) =>
+			checked(() => {
+				const value = object(name);
+				return {
+					ContentLength: value.body.length,
+					ETag: value.identity,
+					Metadata: value.metadata,
+					...(sample.provider === "googleDrive"
+						? {
+								RecordingContentETag: value.identity,
+								RecordingContentSHA256: hash(value.body),
+							}
+						: {}),
+				};
+			}),
+		getObject: (name: string) =>
+			checked(() => {
+				if (!name.endsWith(".json"))
+					throw new Error("Media bytes entered the control plane");
+				return Option.fromNullable(objects.get(name)?.body.toString("utf8"));
+			}),
+		putObject: (name: string, body: string) =>
+			checked(() => {
+				if (
+					!name.startsWith(`${prefix}/.recording/`) ||
+					!name.endsWith(".json")
+				)
+					throw new Error("Unexpected source write");
+				const bytes = Buffer.from(body);
+				objects.set(name, { body: bytes, identity: `"${hash(bytes)}"` });
+			}),
+		copyObject: vi.fn(
+			(
+				source: string,
+				target: string,
+				options: {
+					CopySourceIfMatch: string;
+					Metadata: Record<string, string>;
+				},
+			) =>
+				checked(() => {
+					if (
+						!source.startsWith("replay/") ||
+						!target.startsWith(`${prefix}/.recording/`) ||
+						objects.has(target)
+					)
+						throw new Error("Unexpected source copy");
+					const original = object(source.slice("replay/".length));
+					if (original.identity !== options.CopySourceIfMatch)
+						throw new Error("Source changed");
+					objects.set(target, {
+						...original,
+						identity: `"${hash(original.body)}"`,
+						metadata: options.Metadata,
+					});
+				}),
+		),
+		listObjects: ({
+			prefix: filter,
+			maxKeys,
+			continuationToken,
+		}: {
+			prefix: string;
+			maxKeys: number;
+			continuationToken?: string;
+		}) =>
+			checked(() => {
+				const names = [...objects.keys()]
+					.filter((name) => name.startsWith(filter))
+					.sort();
+				const start = Number(continuationToken ?? 0);
+				return {
+					Contents: names.slice(start, start + maxKeys).map((Key) => ({ Key })),
+					IsTruncated: start + maxKeys < names.length,
+					NextContinuationToken: String(start + maxKeys),
+				};
+			}),
+		getInternalSignedObjectUrl: (name: string) =>
+			checked(() => `https://replay.invalid/${name}`),
+	};
+	mocks.storage.mockReturnValue(Effect.succeed([bucket]));
+	const manifest = (complete: boolean) => {
+		const parsed = readCompletedRecordingManifest(
+			JSON.stringify(sample.manifest),
+		);
+		const body = Buffer.from(
+			JSON.stringify({
+				...sample.manifest,
+				is_complete: complete,
+			}),
+		);
+		objects.set(`${prefix}/segments/manifest.json`, {
+			body,
+			identity: `"${hash(body)}"`,
+		});
+		return {
+			version: 1 as const,
+			artifact: { kind: "segments" as const, manifestSha256: hash(body) },
+			requiredAudio: parsed.hasAudio,
+		};
+	};
+	return { objects, object, bucket, manifest };
+}
+
+function isComplete(sample: Sample) {
+	const manifest = readCompletedRecordingManifest(
+		JSON.stringify(sample.manifest),
+	);
+	return (
+		sample.fragments.length ===
+		manifest.videoSegments.length +
+			manifest.audioSegments.length +
+			1 +
+			Number(manifest.hasAudio)
+	);
+}
+
+describe("opt-in replay of private production source copies", () => {
+	it.skipIf(!corpusRoot)("contains a varied captured corpus", () => {
+		expect(corpus.length).toBeGreaterThanOrEqual(2);
+	});
+
+	it.each(corpus)(
+		"preserves every captured byte from $provider source $label",
+		async (sample) => {
+			const storage = replay(sample);
+			storage.manifest(false);
+			const fragments = sample.fragments
+				.filter((fragment) => fragment.index > 0)
+				.map(({ track, index }) => ({ track, index }));
+			for (let offset = 0; offset < fragments.length; offset += 32) {
+				const batch = fragments.slice(offset, offset + 32);
+				expect(
+					await prepareDesktopRecordingSegments(video, batch, async () => true),
+				).toEqual(batch);
+			}
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(fragments.length);
+			await expect(
+				commitDesktopRecordingSource(video, "stopped-too-early"),
+			).rejects.toThrow("incomplete");
+			const verification = storage.manifest(true);
+			if (!isComplete(sample)) {
+				await expect(
+					commitDesktopRecordingSource(
+						video,
+						"missing-late-upload",
+						verification,
+					),
+				).rejects.toThrow("missing");
+				return;
+			}
+			const source = await commitDesktopRecordingSource(
+				video,
+				"replay-generation",
+				verification,
+			);
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(
+				sample.fragments.length,
+			);
+			const urls = await buildDesktopRecordingSourceUrls(video, source);
+			expect(urls.sourceObjects).toHaveLength(sample.fragments.length);
+			const directory = join(corpusRoot ?? "", sample.label, "replay");
+			mkdirSync(directory, { recursive: true });
+			for (const track of ["video", "audio"] as const) {
+				const originals = sample.fragments
+					.filter((fragment) => fragment.track === track)
+					.sort((left, right) => left.index - right.index);
+				if (!originals.length) continue;
+				const trackUrls =
+					track === "video"
+						? [urls.videoInitUrl, ...urls.videoSegmentUrls]
+						: [urls.audioInitUrl, ...urls.audioSegmentUrls];
+				const copied = trackUrls.map((url, index) => {
+					if (!url) throw new Error("Missing committed source URL");
+					const bytes = storage.object(new URL(url).pathname.slice(1)).body;
+					expect(hash(bytes)).toBe(originals[index]?.sha256);
+					return bytes;
+				});
+				writeFileSync(join(directory, `${track}.mp4`), Buffer.concat(copied));
+			}
+			for (const fragment of sample.fragments)
+				expect(
+					hash(storage.object(key(fragment.track, fragment.index)).body),
+				).toBe(fragment.sha256);
+		},
+		120_000,
+	);
+});
diff --git a/apps/web/__tests__/unit/desktop-recording-source.test.ts b/apps/web/__tests__/unit/desktop-recording-source.test.ts
index dd54888689..023f65f502 100644
--- a/apps/web/__tests__/unit/desktop-recording-source.test.ts
+++ b/apps/web/__tests__/unit/desktop-recording-source.test.ts
@@ -26,6 +26,7 @@ import {
 	commitDesktopRecordingSource,
 	DesktopRecordingSourceError,
 	getDesktopRecordingOutputKey,
+	prepareDesktopRecordingSegments,
 } from "@/lib/desktop-recording-source";
 
 type VideoRow = Parameters<typeof commitDesktopRecordingSource>[0];
@@ -423,6 +424,302 @@ beforeEach(() => {
 	vi.clearAllMocks();
 });
 
+describe("recording source preparation during upload", () => {
+	it.each(["missing", "corrupt", "unknown-version"])(
+		"keeps normal finalization available with a %s preparation marker",
+		async (mutation) => {
+			const storage = storageFixture();
+			seedSegments(storage, [1], false);
+			await prepareDesktopRecordingSegments(
+				recording(),
+				[{ track: "video", index: 1 }],
+				async () => true,
+			);
+			const marker = `${prefix}/.recording/sources/preparation.json`;
+			if (mutation === "missing") storage.objects.delete(marker);
+			else
+				storage.seed(
+					marker,
+					mutation === "corrupt" ? "invalid" : JSON.stringify({ version: 2 }),
+				);
+			const source = await commitDesktopRecordingSource(
+				recording(),
+				generation,
+			);
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(3);
+			expect(
+				readInventory(storage, source.inventoryKey).objects.every(
+					(entry) => !entry.key.includes("/prepared/"),
+				),
+			).toBe(true);
+			expect(
+				storage.object(`${prefix}/segments/video/segment_001.m4s`).body,
+			).toBe("video-fragment-1");
+		},
+	);
+
+	it("bounds a stalled recording-state check before copying fragments", async () => {
+		vi.useFakeTimers();
+		try {
+			const storage = storageFixture();
+			seedSegments(storage, [1], false);
+			const result = prepareDesktopRecordingSegments(
+				recording(),
+				[{ track: "video", index: 1 }],
+				() => new Promise<boolean>(() => {}),
+			);
+			const rejected = expect(result).rejects.toThrow("timed out");
+			await vi.advanceTimersByTimeAsync(5_001);
+			await rejected;
+			expect(storage.bucket.copyObject).not.toHaveBeenCalled();
+		} finally {
+			vi.useRealTimers();
+		}
+	});
+
+	it("does not acknowledge stalled storage or hold the request indefinitely", async () => {
+		vi.useFakeTimers();
+		try {
+			const storage = storageFixture();
+			seedSegments(storage, [1, 2, 3, 4, 5], false);
+			storage.bucket.headObject.mockImplementation(() => Effect.never);
+			const result = prepareDesktopRecordingSegments(
+				recording(),
+				Array.from({ length: 5 }, (_, index) => ({
+					track: "video" as const,
+					index: index + 1,
+				})),
+				async () => true,
+			);
+			await vi.advanceTimersByTimeAsync(15_001);
+			expect(await result).toEqual([]);
+			expect(storage.bucket.copyObject).not.toHaveBeenCalled();
+		} finally {
+			vi.useRealTimers();
+		}
+	});
+
+	for (const provider of ["s3", "googleDrive"]) {
+		it(`reuses ${provider} copies after Stop without copying media twice`, async () => {
+			const storage = storageFixture(provider);
+			const { manifest } = seedSegments(storage);
+			storage.seed(
+				manifestKey,
+				JSON.stringify({ ...manifest, is_complete: false }),
+			);
+			const requested = [
+				{ track: "video" as const, index: 1 },
+				{ track: "video" as const, index: 2 },
+				{ track: "audio" as const, index: 1 },
+			];
+			expect(
+				await prepareDesktopRecordingSegments(
+					recording(),
+					requested,
+					async () => true,
+				),
+			).toEqual(requested);
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(3);
+			await prepareDesktopRecordingSegments(
+				recording(),
+				[...requested, { track: "video", index: 1 }],
+				async () => true,
+			);
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(3);
+			storage.seed(
+				manifestKey,
+				JSON.stringify({
+					...manifest,
+				}),
+			);
+			const source = await commitDesktopRecordingSource(
+				recording(),
+				generation,
+			);
+			expect(storage.bucket.copyObject).toHaveBeenCalledTimes(5);
+			const inventory = readInventory(storage, source.inventoryKey);
+			expect(
+				inventory.objects.filter((entry) => entry.key.includes("/prepared/")),
+			).toHaveLength(3);
+			expect(storage.object(videoInitKey).body).toBe("video-init");
+			const urls = await buildDesktopRecordingSourceUrls(recording(), source);
+			expect(urls.videoSegmentUrls).toHaveLength(2);
+		});
+	}
+
+	it("leaves late fragments for normal finalization", async () => {
+		const storage = storageFixture();
+		const { manifest } = seedSegments(storage, [1], false);
+		const prepared = await prepareDesktopRecordingSegments(
+			recording(),
+			[
+				{ track: "video", index: 1 },
+				{ track: "video", index: 2 },
+			],
+			async () => true,
+		);
+		expect(prepared).toEqual([{ track: "video", index: 1 }]);
+		storage.seed(`${prefix}/segments/video/segment_002.m4s`, "late-fragment");
+		storage.seed(
+			manifestKey,
+			JSON.stringify({
+				...manifest,
+				video_segments: [1, 2],
+			}),
+		);
+		const source = await commitDesktopRecordingSource(recording(), generation);
+		expect(readInventory(storage, source.inventoryKey).objects).toHaveLength(3);
+		expect(storage.bucket.copyObject).toHaveBeenCalledTimes(3);
+	});
+
+	it("does not accept missing final fragments because earlier preparation succeeded", async () => {
+		const storage = storageFixture();
+		const { manifest } = seedSegments(storage, [1], false);
+		await prepareDesktopRecordingSegments(
+			recording(),
+			[{ track: "video", index: 1 }],
+			async () => true,
+		);
+		storage.seed(
+			manifestKey,
+			JSON.stringify({
+				...manifest,
+				video_segments: [1, 2],
+			}),
+		);
+		await expect(
+			commitDesktopRecordingSource(recording(), generation),
+		).rejects.toThrow("missing");
+		expect(
+			storage.object(`${prefix}/segments/video/segment_001.m4s`).body,
+		).toBe("video-fragment-1");
+	});
+
+	it("uses the current source when a fragment was uploaded again after preparation", async () => {
+		const storage = storageFixture();
+		const { manifest } = seedSegments(storage, [1], false);
+		await prepareDesktopRecordingSegments(
+			recording(),
+			[{ track: "video", index: 1 }],
+			async () => true,
+		);
+		storage.seed(
+			`${prefix}/segments/video/segment_001.m4s`,
+			"replacement-fragment",
+		);
+		storage.seed(
+			manifestKey,
+			JSON.stringify({
+				...manifest,
+			}),
+		);
+		const source = await commitDesktopRecordingSource(recording(), generation);
+		const output = readInventory(storage, source.inventoryKey).objects;
+		const fragment = output.find((entry) => entry.index === 1);
+		if (!fragment) throw new Error("Missing copied video fragment");
+		expect(output.every((entry) => !entry.key.includes("/prepared/"))).toBe(
+			true,
+		);
+		expect(storage.object(fragment.key).body).toBe("replacement-fragment");
+	});
+
+	for (const mutation of ["receipt", "object", "missing"] as const) {
+		it(`falls back when the optional prepared ${mutation} is invalid`, async () => {
+			const storage = storageFixture();
+			const { manifest } = seedSegments(storage, [1], false);
+			await prepareDesktopRecordingSegments(
+				recording(),
+				[{ track: "video", index: 1 }],
+				async () => true,
+			);
+			const copy = storage.bucket.copyObject.mock.calls[0];
+			if (!copy) throw new Error("Missing prepared copy");
+			const key = copy[1];
+			if (mutation === "receipt")
+				storage.seed(
+					key.replace(/\/video\/1\.m4s$/, "/receipt.json"),
+					"invalid-json",
+				);
+			if (mutation === "object") storage.seed(key, "wrong-fragment");
+			if (mutation === "missing") storage.objects.delete(key);
+			storage.seed(
+				manifestKey,
+				JSON.stringify({
+					...manifest,
+				}),
+			);
+			const source = await commitDesktopRecordingSource(
+				recording(),
+				generation,
+			);
+			const objects = readInventory(storage, source.inventoryKey).objects;
+			const fragment = objects.find((entry) => entry.index === 1);
+			if (!fragment) throw new Error("Missing copied video fragment");
+			expect(objects.every((entry) => !entry.key.includes("/prepared/"))).toBe(
+				true,
+			);
+			expect(storage.object(fragment.key).body).toBe("video-fragment-1");
+		});
+	}
+
+	it("stops scheduling copies when the recording is stopped or deleted", async () => {
+		const storage = storageFixture();
+		seedSegments(storage, [1, 2, 3, 4, 5, 6], false);
+		const canContinue = vi
+			.fn()
+			.mockResolvedValueOnce(true)
+			.mockResolvedValue(false);
+		const prepared = await prepareDesktopRecordingSegments(
+			recording(),
+			Array.from({ length: 6 }, (_, index) => ({
+				track: "video" as const,
+				index: index + 1,
+			})),
+			canContinue,
+		);
+		expect(prepared).toHaveLength(4);
+		expect(storage.bucket.copyObject).toHaveBeenCalledTimes(4);
+		await prepareDesktopRecordingSegments(
+			recording(),
+			[{ track: "video", index: 5 }],
+			async () => false,
+		);
+		expect(storage.bucket.copyObject).toHaveBeenCalledTimes(4);
+	});
+
+	it("bounds speculative copies and rejects init files and oversized batches", async () => {
+		const storage = storageFixture();
+		seedSegments(storage, [1], false);
+		storage.object(`${prefix}/segments/video/segment_001.m4s`).size =
+			65 * 1024 * 1024;
+		expect(
+			await prepareDesktopRecordingSegments(
+				recording(),
+				[{ track: "video", index: 1 }],
+				async () => true,
+			),
+		).toEqual([]);
+		await expect(
+			prepareDesktopRecordingSegments(
+				recording(),
+				[{ track: "video", index: 0 }],
+				async () => true,
+			),
+		).rejects.toThrow();
+		await expect(
+			prepareDesktopRecordingSegments(
+				recording(),
+				Array.from({ length: 33 }, (_, index) => ({
+					track: "video" as const,
+					index: index + 1,
+				})),
+				async () => true,
+			),
+		).rejects.toThrow();
+		expect(storage.bucket.copyObject).not.toHaveBeenCalled();
+	});
+});
+
 describe("durable desktop recording source commit", () => {
 	it("reuses a checksum-verified Drive copy when checksum readiness delayed its receipt", async () => {
 		const storage = storageFixture("googleDrive");
diff --git a/apps/web/__tests__/unit/recording-prepare.test.ts b/apps/web/__tests__/unit/recording-prepare.test.ts
new file mode 100644
index 0000000000..48edfc9c58
--- /dev/null
+++ b/apps/web/__tests__/unit/recording-prepare.test.ts
@@ -0,0 +1,196 @@
+import type { HttpApi } from "@effect/platform";
+import { MySqlDialect } from "drizzle-orm/mysql-core";
+import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+	rows: [] as unknown[][],
+	userId: "owner" as string | null,
+	where: vi.fn(),
+	prepare: vi.fn(),
+	dispose: async () => {},
+}));
+
+vi.mock("@/lib/desktop-recording-source", () => ({
+	prepareDesktopRecordingSegments: mocks.prepare,
+}));
+
+vi.mock("@/lib/server", async () => {
+	const { Database } = await import("@cap/web-backend");
+	const { HttpAuthMiddleware, Organisation, User, DatabaseError } =
+		await import("@cap/web-domain");
+	const { HttpApiBuilder, HttpApiError, HttpServer } = await import(
+		"@effect/platform"
+	);
+	const { Effect, Layer, Option } = await import("effect");
+	const client = {
+		select: () => ({
+			from: () => ({ leftJoin: () => ({ where: mocks.where }) }),
+		}),
+	} as unknown as import("@cap/web-backend/src/Database").DbClient;
+	const database = Layer.succeed(
+		Database,
+		Database.make({
+			use: (callback) =>
+				Effect.tryPromise({
+					try: () => callback(client),
+					catch: (cause) => new DatabaseError({ cause }),
+				}),
+		}),
+	);
+	const auth = Layer.succeed(
+		HttpAuthMiddleware,
+		Effect.suspend(() =>
+			mocks.userId
+				? Effect.succeed({
+						id: User.UserId.make(mocks.userId),
+						email: "owner@example.com",
+						activeOrganizationId: Organisation.OrganisationId.make("org"),
+						iconUrlOrKey: Option.none(),
+					})
+				: Effect.fail(new HttpApiError.Unauthorized()),
+		),
+	);
+	return {
+		apiToHandler: (
+			api: import("effect").Layer.Layer<
+				HttpApi.Api,
+				never,
+				| import("@cap/web-backend").Database
+				| import("@cap/web-domain").HttpAuthMiddleware
+			>,
+		) => {
+			const handler = api.pipe(
+				Layer.provideMerge(auth),
+				Layer.provideMerge(database),
+				Layer.merge(HttpServer.layerContext),
+				HttpApiBuilder.toWebHandler,
+			);
+			mocks.dispose = handler.dispose;
+			return handler.handler;
+		},
+	};
+});
+
+import { POST } from "@/app/api/recording/prepare/route";
+
+const segments = [{ track: "video", index: 1 }];
+const current = {
+	video: {
+		id: "recording",
+		ownerId: "owner",
+		source: { type: "desktopSegments" },
+	},
+	jobId: null,
+};
+
+function request(body: unknown = { videoId: "recording", segments }) {
+	return POST(
+		new Request("http://localhost/api/recording/prepare", {
+			method: "POST",
+			headers: { "Content-Type": "application/json" },
+			body: JSON.stringify(body),
+		}),
+	);
+}
+
+describe("optional recording preparation API", () => {
+	beforeEach(() => {
+		mocks.userId = "owner";
+		mocks.rows = [[current], [current]];
+		mocks.where.mockImplementation(async () => mocks.rows.shift() ?? []);
+		mocks.prepare.mockReset().mockResolvedValue(segments);
+	});
+
+	afterAll(() => mocks.dispose());
+
+	it("requires authentication before accessing storage", async () => {
+		mocks.userId = null;
+		expect((await request()).status).toBe(401);
+		expect(mocks.where).not.toHaveBeenCalled();
+		expect(mocks.prepare).not.toHaveBeenCalled();
+	});
+
+	it("limits the lookup to the authenticated recording owner", async () => {
+		mocks.rows = [[]];
+		expect((await request()).status).toBe(404);
+		const [condition] = mocks.where.mock.calls[0] ?? [];
+		const query = new MySqlDialect().sqlToQuery(condition);
+		expect(query.params).toEqual(["recording", "owner"]);
+		expect(mocks.prepare).not.toHaveBeenCalled();
+	});
+
+	it.each(
+		[
+			[],
+			[{ track: "video", index: 0 }],
+			[{ track: "video", index: 1.5 }],
+			[{ track: "video", index: 50_001 }],
+			[{ track: "../video", index: 1 }],
+			Array.from({ length: 33 }, (_, index) => ({
+				track: "audio",
+				index: index + 1,
+			})),
+		].map((invalid) => ({ invalid })),
+	)(
+		"rejects an invalid or oversized fragment batch $invalid",
+		async ({ invalid }) => {
+			expect(
+				(await request({ videoId: "recording", segments: invalid })).status,
+			).toBe(400);
+			expect(mocks.prepare).not.toHaveBeenCalled();
+		},
+	);
+
+	it.each(["desktopMP4", "webMP4", "local"])(
+		"does not prepare a %s source",
+		async (type) => {
+			mocks.rows = [
+				{ ...current, video: { ...current.video, source: { type } } },
+			].map((row) => [row]);
+			expect(await (await request()).json()).toEqual({
+				version: 1,
+				prepared: [],
+			});
+			expect(mocks.prepare).not.toHaveBeenCalled();
+		},
+	);
+
+	it("does not race an existing finalization or deletion job", async () => {
+		mocks.rows = [[{ ...current, jobId: "recording" }]];
+		expect(await (await request()).json()).toEqual({
+			version: 1,
+			prepared: [],
+		});
+		expect(mocks.prepare).not.toHaveBeenCalled();
+	});
+
+	it("rechecks ownership and finalization between copy batches", async () => {
+		mocks.rows = [
+			[current],
+			[current],
+			[{ ...current, jobId: "recording" }],
+			[],
+		];
+		mocks.prepare.mockImplementation(async (_video, requested, canContinue) => {
+			expect(await canContinue()).toBe(true);
+			expect(await canContinue()).toBe(false);
+			expect(await canContinue()).toBe(false);
+			return requested;
+		});
+		expect(await (await request()).json()).toEqual({
+			version: 1,
+			prepared: segments,
+		});
+		for (const [condition] of mocks.where.mock.calls) {
+			expect(new MySqlDialect().sqlToQuery(condition).params).toEqual([
+				"recording",
+				"owner",
+			]);
+		}
+	});
+
+	it("leaves preparation retryable after a storage error", async () => {
+		mocks.prepare.mockRejectedValue(new Error("Storage unavailable"));
+		expect((await request()).status).toBe(500);
+	});
+});
diff --git a/apps/web/__tests__/unit/s3-bucket-connections.test.ts b/apps/web/__tests__/unit/s3-bucket-connections.test.ts
new file mode 100644
index 0000000000..07df81095f
--- /dev/null
+++ b/apps/web/__tests__/unit/s3-bucket-connections.test.ts
@@ -0,0 +1,230 @@
+import { createServer, request as httpRequest } from "node:http";
+import type { Socket } from "node:net";
+import { S3Bucket } from "@cap/web-domain";
+import { ConfigProvider, Effect, Layer, ManagedRuntime, Option } from "effect";
+import { describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({ getById: vi.fn() }));
+
+vi.mock("@cap/database/crypto", () => ({
+	decrypt: async (value: string) => value,
+}));
+vi.mock("@cap/web-backend/src/Database.ts", async () => {
+	const { Effect } = await import("effect");
+	class Database extends Effect.Service<Database>()("Database", {
+		sync: () => ({}),
+	}) {}
+	return { Database };
+});
+vi.mock("@cap/web-backend/src/Aws.ts", async () => {
+	const { Effect } = await import("effect");
+	class AwsCredentials extends Effect.Service<AwsCredentials>()(
+		"AwsCredentials",
+		{
+			sync: () => ({
+				credentials: {
+					accessKeyId: "default-key",
+					secretAccessKey: "test-secret",
+				},
+			}),
+		},
+	) {}
+	return { AwsCredentials };
+});
+vi.mock("@cap/web-backend/src/S3Buckets/S3BucketsRepo.ts", async () => {
+	const { Effect } = await import("effect");
+	class S3BucketsRepo extends Effect.Service<S3BucketsRepo>()("S3BucketsRepo", {
+		sync: () => ({ getById: mocks.getById }),
+	}) {}
+	return { S3BucketsRepo };
+});
+
+import { S3Buckets } from "@cap/web-backend/src/S3Buckets";
+import { s3ConnectionPool } from "@cap/web-backend/src/S3Buckets/S3ConnectionPool";
+
+async function storageFixture() {
+	let connections = 0;
+	const sockets = new Set<Socket>();
+	const authorizations: string[] = [];
+	const server = createServer((request, response) => {
+		authorizations.push(request.headers.authorization ?? "");
+		response.writeHead(200, {
+			"Content-Length": "1",
+			ETag: '"source-identity"',
+		});
+		response.end();
+	});
+	server.on("connection", (socket) => {
+		connections++;
+		sockets.add(socket);
+		socket.on("close", () => sockets.delete(socket));
+	});
+	await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+	const address = server.address();
+	if (!address || typeof address === "string") throw new Error("Missing port");
+	const endpoint = `http://127.0.0.1:${address.port}`;
+	const runtime = ManagedRuntime.make(
+		S3Buckets.Default.pipe(
+			Layer.provide(
+				Layer.setConfigProvider(
+					ConfigProvider.fromMap(
+						new Map([
+							["CAP_AWS_REGION", "us-east-1"],
+							["CAP_AWS_BUCKET", "capso"],
+							["S3_INTERNAL_ENDPOINT", endpoint],
+							["S3_PUBLIC_ENDPOINT", endpoint],
+						]),
+					),
+				),
+			),
+		),
+	);
+	const service = await runtime.runPromise(S3Buckets);
+	return {
+		endpoint,
+		runtime,
+		service,
+		authorizations,
+		connections: () => connections,
+		async close() {
+			await runtime.dispose();
+			for (const socket of sockets) socket.destroy();
+			await new Promise<void>((resolve, reject) =>
+				server.close((error) => (error ? reject(error) : resolve())),
+			);
+		},
+	};
+}
+
+describe("S3 connection reuse", () => {
+	it("bounds connections across simultaneous recording checkpoints", async () => {
+		const fixture = await storageFixture();
+		try {
+			await Promise.all(
+				Array.from({ length: 24 }, async (_, recording) => {
+					const [access] = await fixture.runtime.runPromise(
+						fixture.service.getBucketAccess(Option.none()),
+					);
+					await Promise.all(
+						Array.from({ length: 8 }, (_, index) =>
+							fixture.runtime.runPromise(
+								access.headObject(`recording/${recording}/${index}`),
+							),
+						),
+					);
+				}),
+			);
+			expect(fixture.authorizations).toHaveLength(192);
+			expect(fixture.connections()).toBeLessThanOrEqual(50);
+		} finally {
+			await fixture.close();
+		}
+	});
+
+	it("expires idle connections, clears their timer on reuse, and closes on disposal", async () => {
+		const fixture = await storageFixture();
+		try {
+			const socket = await Effect.runPromise(
+				Effect.scoped(
+					Effect.gen(function* () {
+						const pool = yield* s3ConnectionPool;
+						return yield* Effect.promise(async () => {
+							const send = () =>
+								new Promise<{ socket: Socket; timeout: number | undefined }>(
+									(resolve, reject) => {
+										let acquired: Socket | undefined;
+										let timeout: number | undefined;
+										const request = httpRequest(fixture.endpoint, {
+											method: "HEAD",
+											agent: pool.httpAgent,
+										});
+										request.on("socket", (socket) => {
+											acquired = socket;
+											timeout = socket.timeout;
+										});
+										request.on("response", (response) => {
+											response.on("end", () => {
+												if (acquired) resolve({ socket: acquired, timeout });
+												else reject(new Error("Missing connection"));
+											});
+											response.resume();
+										});
+										request.on("error", reject);
+										request.end();
+									},
+								);
+							const first = await send();
+							await new Promise<void>((resolve) => setImmediate(resolve));
+							expect(first.socket.timeout).toBeGreaterThan(0);
+							expect(first.socket.timeout).toBeLessThanOrEqual(30_000);
+							const second = await send();
+							expect(second.socket).toBe(first.socket);
+							expect(second.timeout).toBe(0);
+							return first.socket;
+						});
+					}),
+				),
+			);
+			expect(socket.destroyed).toBe(true);
+		} finally {
+			await fixture.close();
+		}
+	});
+
+	it("does not grow connections with every recording checkpoint", async () => {
+		const fixture = await storageFixture();
+		try {
+			for (let checkpoint = 0; checkpoint < 40; checkpoint++) {
+				const [access] = await fixture.runtime.runPromise(
+					fixture.service.getBucketAccess(Option.none()),
+				);
+				await Promise.all(
+					Array.from({ length: 8 }, (_, index) =>
+						fixture.runtime.runPromise(
+							access.headObject(`recording/${checkpoint}/${index}`),
+						),
+					),
+				);
+			}
+			expect(fixture.authorizations).toHaveLength(320);
+			expect(fixture.connections()).toBeLessThanOrEqual(16);
+		} finally {
+			await fixture.close();
+		}
+	});
+
+	it("reuses custom bucket connections while using current credentials", async () => {
+		const fixture = await storageFixture();
+		try {
+			for (let revision = 0; revision < 40; revision++) {
+				mocks.getById.mockReturnValue(
+					Effect.succeed(
+						Option.some(
+							S3Bucket.decodeSync({
+								id: "custom-bucket",
+								ownerId: "owner",
+								region: "us-east-1",
+								endpoint: fixture.endpoint,
+								name: "custom-storage",
+								accessKeyId: `rotated-key-${revision}`,
+								secretAccessKey: "custom-secret",
+							}),
+						),
+					),
+				);
+				const [access] = await fixture.runtime.runPromise(
+					fixture.service.getBucketAccess(
+						Option.some(S3Bucket.S3BucketId.make("custom-bucket")),
+					),
+				);
+				await fixture.runtime.runPromise(access.headObject("fragment.m4s"));
+				expect(fixture.authorizations.at(-1)).toContain(
+					`Credential=rotated-key-${revision}/`,
+				);
+			}
+			expect(fixture.connections()).toBeLessThanOrEqual(2);
+		} finally {
+			await fixture.close();
+		}
+	});
+});
diff --git a/apps/web/app/api/recording/prepare/route.ts b/apps/web/app/api/recording/prepare/route.ts
new file mode 100644
index 0000000000..23fcbef131
--- /dev/null
+++ b/apps/web/app/api/recording/prepare/route.ts
@@ -0,0 +1,109 @@
+import { videoProcessingJobs, videos } from "@cap/database/schema";
+import { Database } from "@cap/web-backend";
+import { CurrentUser, HttpAuthMiddleware, Video } from "@cap/web-domain";
+import {
+	HttpApi,
+	HttpApiBuilder,
+	HttpApiEndpoint,
+	HttpApiError,
+	HttpApiGroup,
+} from "@effect/platform";
+import { and, eq } from "drizzle-orm";
+import { Effect, Layer, Schema } from "effect";
+import { prepareDesktopRecordingSegments } from "@/lib/desktop-recording-source";
+import { apiToHandler } from "@/lib/server";
+
+const Segment = Schema.Struct({
+	track: Schema.Literal("video", "audio"),
+	index: Schema.Int.pipe(Schema.between(1, 50_000)),
+});
+
+class Api extends HttpApi.make("RecordingPreparationApi").add(
+	HttpApiGroup.make("root").add(
+		HttpApiEndpoint.post("prepare")`/api/recording/prepare`
+			.setPayload(
+				Schema.Struct({
+					videoId: Video.VideoId,
+					segments: Schema.Array(Segment).pipe(
+						Schema.minItems(1),
+						Schema.maxItems(32),
+					),
+				}),
+			)
+			.addSuccess(
+				Schema.Struct({
+					version: Schema.Literal(1),
+					prepared: Schema.Array(Segment),
+				}),
+			)
+			.addError(HttpApiError.NotFound)
+			.addError(HttpApiError.InternalServerError)
+			.middleware(HttpAuthMiddleware),
+	),
+) {}
+
+const ApiLive = HttpApiBuilder.api(Api).pipe(
+	Layer.provide(
+		HttpApiBuilder.group(Api, "root", (handlers) =>
+			handlers.handle("prepare", ({ payload }) =>
+				Effect.gen(function* () {
+					const user = yield* CurrentUser;
+					const database = yield* Database;
+					const prepared = yield* database
+						.use(async (client) => {
+							const read = () =>
+								client
+									.select({ video: videos, jobId: videoProcessingJobs.videoId })
+									.from(videos)
+									.leftJoin(
+										videoProcessingJobs,
+										eq(videoProcessingJobs.videoId, videos.id),
+									)
+									.where(
+										and(
+											eq(videos.id, payload.videoId),
+											eq(videos.ownerId, user.id),
+										),
+									);
+							const [current] = await read();
+							if (!current) return null;
+							if (
+								current.jobId ||
+								current.video.source?.type !== "desktopSegments"
+							)
+								return [];
+							return prepareDesktopRecordingSegments(
+								current.video,
+								payload.segments,
+								async () => {
+									const [latest] = await read();
+									return Boolean(
+										latest &&
+											!latest.jobId &&
+											latest.video.source?.type === "desktopSegments",
+									);
+								},
+							);
+						})
+						.pipe(
+							Effect.catchTag("DatabaseError", () =>
+								Effect.fail(new HttpApiError.InternalServerError()),
+							),
+						);
+					if (prepared === null) return yield* new HttpApiError.NotFound();
+					return { version: 1 as const, prepared };
+				}).pipe(
+					Effect.timeoutFail({
+						duration: "20 seconds",
+						onTimeout: () => new HttpApiError.InternalServerError(),
+					}),
+				),
+			),
+		),
+	),
+);
+
+const handler = apiToHandler(ApiLive);
+
+export const POST = handler;
+export const maxDuration = 30;
diff --git a/apps/web/lib/desktop-recording-source.ts b/apps/web/lib/desktop-recording-source.ts
index 1073e539bb..514421bb07 100644
--- a/apps/web/lib/desktop-recording-source.ts
+++ b/apps/web/lib/desktop-recording-source.ts
@@ -212,6 +212,7 @@ const sourcePlanSchema = z.object({
 	videoCount: z.number().int().nonnegative(),
 	audioCount: z.number().int().nonnegative(),
 	objectCount: z.number().int().positive().max(SOURCE_COMMIT_MAX_OBJECTS),
+	sourcePreparation: z.boolean().optional(),
 	mp4: z
 		.object({
 			originalKey: z.string(),
@@ -542,6 +543,20 @@ async function createSourcePlan(
 			videoCount,
 			audioCount,
 			objectCount: videoCount + audioCount,
+			sourcePreparation: await context
+				.run(
+					context.bucket
+						.getObject(preparationMarkerKey(context.video))
+						.pipe(Effect.timeout("2 seconds")),
+				)
+				.then(
+					(marker) =>
+						Option.isSome(marker) &&
+						z
+							.object({ version: z.literal(1) })
+							.safeParse(JSON.parse(marker.value)).success,
+				)
+				.catch(() => false),
 		};
 	} else {
 		const video = context.video;
@@ -683,7 +698,12 @@ async function checkCopy(
 	key: string,
 	expectedIdentity?: string,
 ): Promise<SourceObject> {
-	assertContextKey(context, key);
+	assertContextKey(
+		key.startsWith(`${preparedSourcePrefix(context.video, original)}/`)
+			? { ...context, prefix: preparedSourcePrefix(context.video, original) }
+			: context,
+		key,
+	);
 	const [head, originalHead] = await Promise.all([
 		context.run(context.bucket.headObject(key)),
 		checkOriginal(context, original),
@@ -818,7 +838,16 @@ async function copySmallObject(
 	context: SourceContext,
 	original: OriginalObject,
 	position: number,
+	prepared = false,
 ) {
+	if (prepared && original.index > 0) {
+		const existing = await reusableCopy(
+			{ ...context, prefix: preparedSourcePrefix(context.video, original) },
+			original,
+			original.index,
+		).catch(() => null);
+		if (existing) return existing;
+	}
 	const existing = await reusableCopy(context, original, position);
 	if (existing) return existing;
 	const key = newCopyKey(context, original, position);
@@ -838,6 +867,109 @@ async function copySmallObject(
 	return saveObjectReceipt(context, await checkCopy(context, original, key));
 }
 
+function preparedSourcePrefix(video: DbVideo, original: OriginalObject) {
+	return `${sourcePrefix(video)}prepared/${hash(
+		JSON.stringify([
+			original.originalKey,
+			original.originalIdentity,
+			original.size,
+		]),
+	)}`;
+}
+
+function preparationMarkerKey(video: DbVideo) {
+	return `${sourcePrefix(video)}preparation.json`;
+}
+
+export const recordingPreparationSegmentSchema = z.object({
+	track: z.enum(["video", "audio"]),
+	index: z.number().int().min(1).max(50_000),
+});
+
+export type RecordingPreparationSegment = z.infer<
+	typeof recordingPreparationSegmentSchema
+>;
+
+export async function prepareDesktopRecordingSegments(
+	video: DbVideo,
+	segments: readonly RecordingPreparationSegment[],
+	canContinue: () => Promise<boolean>,
+): Promise<RecordingPreparationSegment[]> {
+	identifierSchema.parse(video.ownerId);
+	identifierSchema.parse(video.id);
+	const requested = z
+		.array(recordingPreparationSegmentSchema)
+		.min(1)
+		.max(32)
+		.parse(segments);
+	if (video.source?.type !== "desktopSegments") return [];
+	const deadline = Date.now() + 15_000;
+	const run: SourceRun = (operation) => {
+		const remaining = deadline - Date.now();
+		if (remaining <= 0)
+			throw new Error("Recording preparation time budget ended");
+		return runWorkflowPromise(
+			operation.pipe(Effect.timeout(Math.min(5_000, remaining))),
+		);
+	};
+	const [bucket] = await run(
+		Storage.getAccessForVideo(decodeStorageVideo(video), {
+			resolvePublishedOutput: false,
+		}),
+	);
+	const context = {
+		video,
+		bucket,
+		run,
+		prefix: sourcePrefix(video).slice(0, -1),
+	};
+	const unique = [
+		...new Map(
+			requested.map((entry) => [`${entry.track}/${entry.index}`, entry]),
+		).values(),
+	];
+	const prepared: RecordingPreparationSegment[] = [];
+	let markerSaved = false;
+	for (
+		let offset = 0;
+		offset < unique.length && Date.now() < deadline;
+		offset += 4
+	) {
+		if (!(await run(Effect.tryPromise(canContinue)))) break;
+		const results = await Promise.all(
+			unique.slice(offset, offset + 4).map(async (segment) => {
+				try {
+					const original = await captureOriginal(context, {
+						originalKey: `${video.ownerId}/${video.id}/segments/${segment.track}/segment_${String(segment.index).padStart(3, "0")}.m4s`,
+						...segment,
+					});
+					if (original.size > 64 * 1024 * 1024) return null;
+					await copySmallObject(
+						{ ...context, prefix: preparedSourcePrefix(video, original) },
+						original,
+						original.index,
+					);
+					return segment;
+				} catch {
+					return null;
+				}
+			}),
+		);
+		for (const segment of results) {
+			if (segment) prepared.push(segment);
+		}
+		if (!markerSaved && prepared.length > 0) {
+			await writeSourceText(
+				context,
+				preparationMarkerKey(video),
+				JSON.stringify({ version: 1 }),
+			);
+			markerSaved = true;
+		}
+	}
+	return prepared;
+}
+
 async function advanceMultipartCopy(
 	context: SourceContext,
 	checkpoint: DesktopRecordingSourceCheckpoint,
@@ -1097,7 +1229,12 @@ async function advanceSourceSnapshot(
 					position: checkpoint.cursor + index,
 				})),
 				({ original, position }) =>
-					copySmallObject(context, original, position),
+					copySmallObject(
+						context,
+						original,
+						position,
+						plan.sourcePreparation === true,
+					),
 			);
 		}
 		const receiptRoots = await appendTree(
diff --git a/crates/recording/src/lib.rs b/crates/recording/src/lib.rs
index 849f46b50a..adf7be98b2 100644
--- a/crates/recording/src/lib.rs
+++ b/crates/recording/src/lib.rs
@@ -17,6 +17,7 @@ pub mod sources;
 pub mod studio_recording;
 pub mod sync_calibration;
 pub mod track_heal;
+pub mod upload_preparation;
 pub mod upload_resume;
 pub mod upload_verification;
 
diff --git a/crates/recording/src/upload_preparation.rs b/crates/recording/src/upload_preparation.rs
new file mode 100644
index 0000000000..bc74ec5bd0
--- /dev/null
+++ b/crates/recording/src/upload_preparation.rs
@@ -0,0 +1,173 @@
+use serde::{Deserialize, Serialize};
+use std::collections::{BTreeMap, BTreeSet};
+use std::time::Duration;
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Track {
+    Video,
+    Audio,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
+pub struct Segment {
+    pub track: Track,
+    pub index: u32,
+}
+
+#[derive(Default)]
+pub struct Preparation {
+    attempts: BTreeMap<Segment, u8>,
+    acknowledged: BTreeSet<Segment>,
+    consecutive_failures: u32,
+}
+
+impl Preparation {
+    pub fn next_batch(
+        &mut self,
+        video: impl IntoIterator<Item = u32>,
+        audio: impl IntoIterator<Item = u32>,
+    ) -> Vec<Segment> {
+        let mut available: Vec<_> = video
+            .into_iter()
+            .map(|index| Segment {
+                track: Track::Video,
+                index,
+            })
+            .chain(audio.into_iter().map(|index| Segment {
+                track: Track::Audio,
+                index,
+            }))
+            .filter(|segment| {
+                (1..=50_000).contains(&segment.index)
+                    && !self.acknowledged.contains(segment)
+                    && self.attempts.get(segment).copied().unwrap_or(0) < 2
+            })
+            .collect();
+        available.sort_by_key(|segment| (segment.index, segment.track));
+        available.truncate(32);
+        for segment in &available {
+            *self.attempts.entry(*segment).or_default() += 1;
+        }
+        available
+    }
+
+    pub fn acknowledge(&mut self, requested: &[Segment], prepared: &[Segment]) {
+        let before = self.acknowledged.len();
+        self.acknowledged.extend(
+            prepared
+                .iter()
+                .filter(|segment| requested.contains(segment))
+                .copied(),
+        );
+        if self.acknowledged.len() > before {
+            self.consecutive_failures = 0;
+        } else {
+            self.request_failed();
+        }
+    }
+
+    pub fn request_failed(&mut self) {
+        self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+    }
+
+    pub fn retry_delay(&self) -> Duration {
+        if self.consecutive_failures == 0 {
+            return Duration::ZERO;
+        }
+        Duration::from_secs((30u64 << self.consecutive_failures.min(4)).min(300))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn batches_only_uploaded_fragments_with_equal_track_priority() {
+        let batch = Preparation::default().next_batch((1..=40).filter(|index| *index != 3), 1..=40);
+        assert_eq!(batch.len(), 32);
+        assert!(batch.contains(&Segment {
+            track: Track::Audio,
+            index: 3
+        }));
+        assert!(!batch.contains(&Segment {
+            track: Track::Video,
+            index: 3
+        }));
+        assert!(
+            batch
+                .iter()
+                .filter(|item| item.track == Track::Video)
+                .count()
+                >= 15
+        );
+        assert!(
+            Preparation::default()
+                .next_batch([0, 50_001], [])
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn bounds_lost_responses_and_does_not_repeat_acknowledged_fragments() {
+        let mut preparation = Preparation::default();
+        let batch = preparation.next_batch([1, 2], [1]);
+        preparation.acknowledge(
+            &batch,
+            &[
+                Segment {
+                    track: Track::Audio,
+                    index: 1,
+                },
+                Segment {
+                    track: Track::Video,
+                    index: 99,
+                },
+            ],
+        );
+        let retry = preparation.next_batch([1, 2, 99], [1]);
+        assert_eq!(retry.len(), 3);
+        preparation.acknowledge(&retry, &retry);
+        assert!(preparation.next_batch([1, 2, 99], [1]).is_empty());
+    }
+
+    #[test]
+    fn an_exhausted_fragment_does_not_disable_later_uploads() {
+        let mut preparation = Preparation::default();
+        assert_eq!(preparation.next_batch([1], []).len(), 1);
+        assert_eq!(preparation.next_batch([1], []).len(), 1);
+        assert_eq!(
+            preparation.next_batch([1, 2], []),
+            vec![Segment {
+                track: Track::Video,
+                index: 2
+            }]
+        );
+    }
+
+    #[test]
+    fn backs_off_outages_and_resets_after_real_progress() {
+        let mut preparation = Preparation::default();
+        let batch = preparation.next_batch([1], []);
+        preparation.acknowledge(&batch, &[]);
+        assert_eq!(preparation.retry_delay(), Duration::from_secs(60));
+        preparation.request_failed();
+        assert_eq!(preparation.retry_delay(), Duration::from_secs(120));
+        for _ in 0..10 {
+            preparation.request_failed();
+        }
+        assert_eq!(preparation.retry_delay(), Duration::from_secs(300));
+        preparation.acknowledge(&batch, &batch);
+        assert_eq!(preparation.retry_delay(), Duration::ZERO);
+    }
+
+    #[test]
+    fn preserves_wire_format_without_initialization_fragments() {
+        let batch = Preparation::default().next_batch([0, 1], []);
+        assert_eq!(
+            serde_json::to_value(batch).unwrap(),
+            serde_json::json!([{"track": "video", "index": 1}])
+        );
+    }
+}
diff --git a/packages/web-backend/src/S3Buckets/S3ConnectionPool.ts b/packages/web-backend/src/S3Buckets/S3ConnectionPool.ts
new file mode 100644
index 0000000000..be0a15c8a3
--- /dev/null
+++ b/packages/web-backend/src/S3Buckets/S3ConnectionPool.ts
@@ -0,0 +1,41 @@
+import { Agent as HttpAgent } from "node:http";
+import { Agent as HttpsAgent } from "node:https";
+import { Socket } from "node:net";
+import { Effect } from "effect";
+
+function boundIdleConnections<T extends HttpAgent>(agent: T) {
+	// Idle custom hosts must release their slots without timing out an active copy.
+	const keepSocketAlive = agent.keepSocketAlive.bind(agent);
+	const reuseSocket = agent.reuseSocket.bind(agent);
+	agent.keepSocketAlive = (socket) => {
+		const keep: unknown = keepSocketAlive(socket);
+		if (keep !== false && socket instanceof Socket)
+			socket.setTimeout(Math.min(socket.timeout || 30_000, 30_000));
+		return keep;
+	};
+	agent.reuseSocket = (socket, request) => {
+		if (socket instanceof Socket) socket.setTimeout(0);
+		reuseSocket(socket, request);
+	};
+	return agent;
+}
+
+export const s3ConnectionPool = Effect.acquireRelease(
+	Effect.sync(() => {
+		const options = {
+			keepAlive: true,
+			maxSockets: 50,
+			maxTotalSockets: 128,
+			maxFreeSockets: 8,
+		};
+		return {
+			httpAgent: boundIdleConnections(new HttpAgent(options)),
+			httpsAgent: boundIdleConnections(new HttpsAgent(options)),
+		};
+	}),
+	(pool) =>
+		Effect.sync(() => {
+			pool.httpAgent.destroy();
+			pool.httpsAgent.destroy();
+		}),
+);
diff --git a/packages/web-backend/src/S3Buckets/index.ts b/packages/web-backend/src/S3Buckets/index.ts
index 64a6c481bc..49ecb912ec 100644
--- a/packages/web-backend/src/S3Buckets/index.ts
+++ b/packages/web-backend/src/S3Buckets/index.ts
@@ -10,11 +10,13 @@ import { Database } from "../Database.ts";
 import { createS3BucketAccess } from "./S3BucketAccess.ts";
 import { S3BucketClientProvider } from "./S3BucketClientProvider.ts";
 import { S3BucketsRepo } from "./S3BucketsRepo.ts";
+import { s3ConnectionPool } from "./S3ConnectionPool.ts";
 
 export class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
-	effect: Effect.gen(function* () {
+	scoped: Effect.gen(function* () {
 		const repo = yield* S3BucketsRepo;
 		const { credentials } = yield* AwsCredentials;
+		const requestHandler = yield* s3ConnectionPool;
 
 		const defaultConfigs = {
 			publicEndpoint: yield* Config.string("S3_PUBLIC_ENDPOINT").pipe(
@@ -43,7 +45,10 @@ export class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
 				credentials: defaultConfigs.credentials,
 				forcePathStyle: defaultConfigs.forcePathStyle,
 				requestStreamBufferSize: 16 * 1024,
+				requestHandler,
 			});
+		const defaultInternalClient = createDefaultClient(true);
+		const defaultPublicClient = createDefaultClient(false);
 
 		const endpointIsPathStyle = (endpoint: string, bucket: string) => {
 			try {
@@ -78,6 +83,7 @@ export class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
 						Option.getOrNull,
 					) ?? true,
 				useArnRegion: false,
+				requestHandler,
 			});
 		};
 
@@ -142,8 +148,8 @@ export class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
 			const bucketAccess = yield* Option.match(customBucket, {
 				onNone: () => {
 					const provider = Layer.succeed(S3BucketClientProvider, {
-						getInternal: Effect.succeed(createDefaultClient(true)),
-						getPublic: Effect.succeed(createDefaultClient(false)),
+						getInternal: Effect.succeed(defaultInternalClient),
+						getPublic: Effect.succeed(defaultPublicClient),
 						bucket: defaultConfigs.bucket,
 						isPathStyle: defaultConfigs.forcePathStyle,
 					});