diff --git a/CHANGELOG.md b/CHANGELOG.md
index eea696893..b78cb4201 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,12 @@
 - Display the analyzed song tempo (BPM) as a badge in the rehearsal workspace.
 - 각 합주 역할(Role)별 개인 연습 진행도를 0~100% 범위로 기록 및 시각화할 수 있는 연습 진척도(`practiceProgress`) 트래커 기능 추가. UI 컨트롤(슬라이더 및 +/- 버튼)과 한/영 다국어 지원 포함.
 
+### Fixed
+
+- Enforce one canonical local-audio resource policy across native local-file/YouTube bootstrap intake, the desktop bridge, Python request preflight, temporal decoding, and stem separation so oversized, overlong, malformed, wrong-rate, or non-finite input fails before bootstrap storage or expensive analysis/model work.
+- Align YouTube download admission with that same 100 MiB encoded-byte ceiling: abort in-flight with yt-dlp `max_filesize` and a progress hook, reject announced oversize before `download=True`, delete owned `.part` / `.ytdl` / `-Frag*` siblings from that import directory on abort, and delete post-download artifacts that still exceed the policy. A 60 MiB import that the old 50 MB check rejected is now accepted; a file one byte over 100 MiB is not.
+- Bound native stored-score PDF reads to the 25 MiB product limit before heap allocation and revalidate PDF magic on the same opened descriptor, preventing an attached score that later grows from bypassing the local resource boundary.
+
 ## [0.1.3] - 2026-04-29
 
 ### Fixed
diff --git a/apps/desktop/core/Cargo.toml b/apps/desktop/core/Cargo.toml
index b01a537dc..44f482e73 100644
--- a/apps/desktop/core/Cargo.toml
+++ b/apps/desktop/core/Cargo.toml
@@ -7,7 +7,7 @@ publish = false
 
 [lib]
 name = "bandscope_desktop_core"
-path = "src/lib.rs"
+path = "src/root.rs"
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage)'] }
diff --git a/apps/desktop/core/src/audio_resource.rs b/apps/desktop/core/src/audio_resource.rs
new file mode 100644
index 000000000..88c07e425
--- /dev/null
+++ b/apps/desktop/core/src/audio_resource.rs
@@ -0,0 +1,22 @@
+/// Maximum encoded local-audio file size accepted by the desktop bootstrap boundary.
+pub const MAX_LOCAL_AUDIO_FILE_BYTES: u64 = 100 * 1024 * 1024;
+
+const LOCAL_AUDIO_READ_ERROR: &str = "Could not read the selected audio file.";
+const LOCAL_AUDIO_TOO_LARGE_ERROR: &str =
+    "Selected audio file exceeds the 100 MiB analysis limit.";
+
+/// Validate a native local-audio file length before storing bootstrap metadata.
+///
+/// The caller must obtain this length from the native filesystem descriptor or
+/// metadata boundary rather than from renderer-controlled JSON. The function
+/// intentionally returns only bounded product messages and never includes a
+/// local path or payload content.
+pub fn validate_local_audio_file_size(file_size_bytes: u64) -> Result<u64, String> {
+    if file_size_bytes == 0 {
+        return Err(LOCAL_AUDIO_READ_ERROR.to_string());
+    }
+    if file_size_bytes > MAX_LOCAL_AUDIO_FILE_BYTES {
+        return Err(LOCAL_AUDIO_TOO_LARGE_ERROR.to_string());
+    }
+    Ok(file_size_bytes)
+}
diff --git a/apps/desktop/core/src/root.rs b/apps/desktop/core/src/root.rs
new file mode 100644
index 000000000..125d13daa
--- /dev/null
+++ b/apps/desktop/core/src/root.rs
@@ -0,0 +1,15 @@
+//! Pure, GUI-independent logic for the BandScope desktop application.
+//!
+//! The historical desktop-core implementation remains in `lib.rs` as the
+//! compatibility module while bounded resource boundaries are isolated in
+//! auditable modules. Public symbols are re-exported so downstream callers keep
+//! the same crate-root API.
+
+#[path = "lib.rs"]
+mod runtime_core;
+mod audio_resource;
+mod score_pdf;
+
+pub use audio_resource::{validate_local_audio_file_size, MAX_LOCAL_AUDIO_FILE_BYTES};
+pub use runtime_core::*;
+pub use score_pdf::read_validated_score_pdf;
diff --git a/apps/desktop/core/src/score_pdf.rs b/apps/desktop/core/src/score_pdf.rs
new file mode 100644
index 000000000..2b26744cc
--- /dev/null
+++ b/apps/desktop/core/src/score_pdf.rs
@@ -0,0 +1,82 @@
+use crate::{MAX_SCORE_PDF_BYTES, PDF_MAGIC};
+use std::{fs::File, io::Read, path::Path};
+
+const SCORE_READ_ERROR: &str = "Could not read the score PDF.";
+const SCORE_TOO_LARGE_ERROR: &str = "Score PDF is too large (exceeds 25MB limit).";
+const SCORE_INVALID_PDF_ERROR: &str = "Stored score is not a valid PDF.";
+
+fn read_validated_pdf_stream(
+    reader: &mut impl Read,
+    expected_len: u64,
+) -> Result<Vec<u8>, String> {
+    if expected_len > MAX_SCORE_PDF_BYTES {
+        return Err(SCORE_TOO_LARGE_ERROR.to_string());
+    }
+
+    // MAX_SCORE_PDF_BYTES is 25 MiB, which fits every supported Rust `usize`.
+    let mut bytes = vec![0_u8; expected_len as usize];
+    reader
+        .read_exact(&mut bytes)
+        .map_err(|_| SCORE_READ_ERROR.to_string())?;
+
+    let mut growth_probe = [0_u8; 1];
+    if reader
+        .read(&mut growth_probe)
+        .map_err(|_| SCORE_READ_ERROR.to_string())?
+        != 0
+    {
+        return Err(SCORE_TOO_LARGE_ERROR.to_string());
+    }
+
+    if !bytes.starts_with(PDF_MAGIC) {
+        return Err(SCORE_INVALID_PDF_ERROR.to_string());
+    }
+
+    Ok(bytes)
+}
+
+/// Read one already-authorized stored score without allocating beyond the PDF limit.
+///
+/// The caller remains responsible for path authority and containment. This helper
+/// opens that resolved path once, snapshots the descriptor length, allocates only
+/// that bounded size, reads exactly that many bytes, and then probes one additional
+/// byte on the same descriptor. A file that was already oversized is rejected
+/// before heap allocation; a file that grows after metadata inspection is rejected
+/// by the one-byte probe without extending the heap buffer beyond the product cap.
+/// Errors intentionally omit the local path and file content.
+pub fn read_validated_score_pdf(path: &Path) -> Result<Vec<u8>, String> {
+    let mut file = File::open(path).map_err(|_| SCORE_READ_ERROR.to_string())?;
+    let metadata = file
+        .metadata()
+        .map_err(|_| SCORE_READ_ERROR.to_string())?;
+    if !metadata.is_file() {
+        return Err(SCORE_READ_ERROR.to_string());
+    }
+    read_validated_pdf_stream(&mut file, metadata.len())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Cursor;
+
+    #[test]
+    fn stream_rejects_growth_after_the_metadata_length_snapshot() {
+        let mut reader = Cursor::new(b"%PDF-extra".to_vec());
+
+        let error = read_validated_pdf_stream(&mut reader, PDF_MAGIC.len() as u64)
+            .expect_err("bytes beyond the metadata snapshot must fail closed");
+
+        assert_eq!(error, SCORE_TOO_LARGE_ERROR);
+    }
+
+    #[test]
+    fn stream_rejects_truncation_after_the_metadata_length_snapshot() {
+        let mut reader = Cursor::new(PDF_MAGIC.to_vec());
+
+        let error = read_validated_pdf_stream(&mut reader, (PDF_MAGIC.len() + 1) as u64)
+            .expect_err("truncation after the metadata snapshot must fail closed");
+
+        assert_eq!(error, SCORE_READ_ERROR);
+    }
+}
diff --git a/apps/desktop/core/tests/audio_resource_policy.rs b/apps/desktop/core/tests/audio_resource_policy.rs
new file mode 100644
index 000000000..11c6725ca
--- /dev/null
+++ b/apps/desktop/core/tests/audio_resource_policy.rs
@@ -0,0 +1,25 @@
+use bandscope_desktop_core::{validate_local_audio_file_size, MAX_LOCAL_AUDIO_FILE_BYTES};
+
+#[test]
+fn local_audio_size_policy_accepts_the_exact_native_bootstrap_ceiling() {
+    assert_eq!(
+        validate_local_audio_file_size(MAX_LOCAL_AUDIO_FILE_BYTES),
+        Ok(MAX_LOCAL_AUDIO_FILE_BYTES)
+    );
+}
+
+#[test]
+fn local_audio_size_policy_rejects_an_empty_native_bootstrap_source() {
+    assert_eq!(
+        validate_local_audio_file_size(0),
+        Err("Could not read the selected audio file.".to_string())
+    );
+}
+
+#[test]
+fn local_audio_size_policy_rejects_a_native_source_above_the_canonical_ceiling() {
+    assert_eq!(
+        validate_local_audio_file_size(MAX_LOCAL_AUDIO_FILE_BYTES + 1),
+        Err("Selected audio file exceeds the 100 MiB analysis limit.".to_string())
+    );
+}
diff --git a/apps/desktop/core/tests/score_pdf_read.rs b/apps/desktop/core/tests/score_pdf_read.rs
new file mode 100644
index 000000000..b70068931
--- /dev/null
+++ b/apps/desktop/core/tests/score_pdf_read.rs
@@ -0,0 +1,91 @@
+use bandscope_desktop_core::{read_validated_score_pdf, MAX_SCORE_PDF_BYTES};
+use std::io::Write;
+use std::path::PathBuf;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+fn unique_test_dir(name: &str) -> PathBuf {
+    let suffix = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("system clock should be after epoch")
+        .as_nanos();
+    std::env::temp_dir().join(format!("bandscope-{name}-{suffix}"))
+}
+
+#[test]
+fn score_pdf_read_returns_only_valid_bounded_pdf_bytes() {
+    let root = unique_test_dir("score-read-valid");
+    std::fs::create_dir_all(&root).expect("test directory should be created");
+    let path = root.join("score.pdf");
+    let expected = b"%PDF-1.7\nvalidated body";
+    std::fs::write(&path, expected).expect("valid PDF fixture should be written");
+
+    let actual = read_validated_score_pdf(&path).expect("valid stored PDF should be readable");
+
+    assert_eq!(actual, expected);
+    let _ = std::fs::remove_dir_all(root);
+}
+
+#[test]
+fn score_pdf_read_rejects_empty_short_and_wrong_magic_content() {
+    let root = unique_test_dir("score-read-invalid");
+    std::fs::create_dir_all(&root).expect("test directory should be created");
+
+    for (name, content) in [
+        ("empty.pdf", b"".as_slice()),
+        ("short.pdf", b"%PD".as_slice()),
+        ("wrong.pdf", b"PK\x03\x04 not a PDF".as_slice()),
+    ] {
+        let path = root.join(name);
+        std::fs::write(&path, content).expect("invalid PDF fixture should be written");
+        let error = read_validated_score_pdf(&path).expect_err("invalid PDF must fail closed");
+        assert!(
+            error == "Could not read the score PDF." || error == "Stored score is not a valid PDF.",
+            "unexpected payload-safe error: {error}"
+        );
+        assert!(!error.contains(root.to_string_lossy().as_ref()));
+    }
+
+    let _ = std::fs::remove_dir_all(root);
+}
+
+#[test]
+fn score_pdf_read_rejects_oversized_sparse_file_before_heap_allocation() {
+    let root = unique_test_dir("score-read-oversized");
+    std::fs::create_dir_all(&root).expect("test directory should be created");
+    let path = root.join("oversized.pdf");
+    let mut file = std::fs::File::create(&path).expect("oversized PDF fixture should be created");
+    file.write_all(b"%PDF-")
+        .expect("PDF magic should be written before extending sparse file");
+    file.set_len(MAX_SCORE_PDF_BYTES + 1)
+        .expect("sparse PDF fixture should exceed the product limit");
+    drop(file);
+
+    let error = read_validated_score_pdf(&path).expect_err("oversized PDF must fail closed");
+
+    assert_eq!(error, "Score PDF is too large (exceeds 25MB limit).");
+    let _ = std::fs::remove_dir_all(root);
+}
+
+#[cfg(unix)]
+#[test]
+fn score_pdf_read_rejects_non_file_descriptor() {
+    let root = unique_test_dir("score-read-directory");
+    std::fs::create_dir_all(&root).expect("test directory should be created");
+
+    let error = read_validated_score_pdf(&root).expect_err("directory must fail closed");
+
+    assert_eq!(error, "Could not read the score PDF.");
+    assert!(!error.contains(root.to_string_lossy().as_ref()));
+    let _ = std::fs::remove_dir_all(root);
+}
+
+#[test]
+fn score_pdf_read_rejects_missing_file_without_exposing_path() {
+    let root = unique_test_dir("score-read-missing");
+    let path = root.join("private-score.pdf");
+
+    let error = read_validated_score_pdf(&path).expect_err("missing PDF must fail closed");
+
+    assert_eq!(error, "Could not read the score PDF.");
+    assert!(!error.contains("private-score.pdf"));
+}
diff --git a/apps/desktop/src-tauri/src/main.rs b/apps/desktop/src-tauri/src/main.rs
index ed4f967bd..1a78bfbf3 100644
--- a/apps/desktop/src-tauri/src/main.rs
+++ b/apps/desktop/src-tauri/src/main.rs
@@ -155,9 +155,10 @@ fn normalize_local_audio_source(path: &Path) -> Result<LocalAudioSourcePayload,
     }
     let metadata = std::fs::metadata(&canonical)
         .map_err(|_| "Could not read the selected audio file.".to_string())?;
-    if !metadata.is_file() || metadata.len() == 0 {
+    if !metadata.is_file() {
         return Err("Could not read the selected audio file.".into());
     }
+    let file_size_bytes = validate_local_audio_file_size(metadata.len())?;
     let file_name = canonical
         .file_name()
         .and_then(|value| value.to_str())
@@ -167,7 +168,7 @@ fn normalize_local_audio_source(path: &Path) -> Result<LocalAudioSourcePayload,
         source_path: canonical.to_string_lossy().into_owned(),
         file_name: file_name.to_string(),
         extension,
-        file_size_bytes: metadata.len(),
+        file_size_bytes,
     })
 }
 
@@ -712,6 +713,7 @@ async fn import_youtube_url(
     if parsed.get("ok").and_then(|v| v.as_bool()) == Some(true) {
         if let Some(metadata) = parsed.get("metadata") {
             let source = youtube_source_from_metadata(metadata, &cache_root)?;
+            validate_local_audio_file_size(source.file_size_bytes)?;
 
             let summary = ProjectBootstrapSummaryPayload {
                 project_id,
@@ -826,7 +828,9 @@ fn attach_score_pdf(
 /// Security Notes: no path crosses the IPC boundary. Both ids are validated
 /// against strict allowlist shapes, the path is rebuilt locally, and the
 /// canonicalize-plus-prefix guard in `resolve_existing_score_pdf` rejects any
-/// escape from the app-owned scores root.
+/// escape from the app-owned scores root. The resolved file is then read
+/// through the bounded core helper so growth after attachment cannot trigger
+/// an allocation beyond the 25 MiB product limit.
 #[tauri::command]
 fn read_score_pdf(
     project_id: String,
@@ -838,7 +842,7 @@ fn read_score_pdf(
     }
     let scores_root = scores_root_for_project(&app, &project_id)?;
     let path = resolve_existing_score_pdf(&scores_root, &score_id)?;
-    std::fs::read(path).map_err(|_| "Could not read the score PDF.".to_string())
+    read_validated_score_pdf(&path)
 }
 
 /// Security Notes: same id validation and traversal guard as `read_score_pdf`;
diff --git a/apps/desktop/src/lib/analysis.resource-policy.test.ts b/apps/desktop/src/lib/analysis.resource-policy.test.ts
new file mode 100644
index 000000000..33ab81804
--- /dev/null
+++ b/apps/desktop/src/lib/analysis.resource-policy.test.ts
@@ -0,0 +1,58 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { importYoutubeUrl, selectLocalAudioSource } from "./analysis";
+
+type TauriWindow = Window & {
+  __TAURI_INTERNALS__?: unknown;
+  __TAURI_INVOKE__?: unknown;
+};
+
+const tauriWindow = window as TauriWindow;
+const INVALID_RESOURCE_POLICY_MESSAGE =
+  "Selected audio file metadata violates the analysis resource policy.";
+
+function fractionalBootstrap(projectId: string) {
+  return {
+    projectId,
+    sourceMode: "reference",
+    projectRoot: `/tmp/bandscope/projects/${projectId}`,
+    cacheRoot: `/tmp/bandscope/cache/${projectId}`,
+    tempRoot: `/tmp/bandscope/temp/${projectId}`,
+    source: {
+      sourcePath: `/tmp/bandscope/${projectId}/input.wav`,
+      fileName: "input.wav",
+      extension: "wav",
+      fileSizeBytes: 1.5
+    }
+  };
+}
+
+describe("analysis encoded-byte policy parity", () => {
+  beforeEach(() => {
+    delete tauriWindow.__TAURI_INTERNALS__;
+    delete tauriWindow.__TAURI_INVOKE__;
+  });
+
+  it("rejects fractional local-file metadata before project state", async () => {
+    tauriWindow.__TAURI_INVOKE__ = vi.fn().mockResolvedValue(fractionalBootstrap("local-project"));
+
+    await expect(selectLocalAudioSource()).resolves.toEqual({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        message: INVALID_RESOURCE_POLICY_MESSAGE
+      }
+    });
+  });
+
+  it("rejects fractional imported-file metadata before project state", async () => {
+    tauriWindow.__TAURI_INVOKE__ = vi.fn().mockResolvedValue(fractionalBootstrap("youtube-project"));
+
+    await expect(importYoutubeUrl("https://youtu.be/4ozX4yFUC34")).resolves.toEqual({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        message: INVALID_RESOURCE_POLICY_MESSAGE
+      }
+    });
+  });
+});
diff --git a/apps/desktop/src/lib/analysis.test.ts b/apps/desktop/src/lib/analysis.test.ts
index e3347d1f5..62170fd23 100644
--- a/apps/desktop/src/lib/analysis.test.ts
+++ b/apps/desktop/src/lib/analysis.test.ts
@@ -1,9 +1,11 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { createDemoAnalysisJobRequest, createDemoRehearsalSong } from "@bandscope/shared-types";
 import {
+  MAX_LOCAL_AUDIO_FILE_BYTES,
   MAX_YOUTUBE_URL_LENGTH,
   getAnalysisJobStatus,
   importYoutubeUrl,
+  selectLocalAudioSource,
   startAnalysisJob
 } from "./analysis";
 
@@ -20,6 +22,58 @@ describe("analysis bridge", () => {
     delete tauriWindow.__TAURI_INVOKE__;
   });
 
+  it("rejects an oversized native local-audio selection before it becomes project state", async () => {
+    tauriWindow.__TAURI_INVOKE__ = vi.fn().mockResolvedValue({
+      projectId: "native-local-project",
+      sourceMode: "reference",
+      projectRoot: "/tmp/bandscope/projects/native-local-project",
+      cacheRoot: "/tmp/bandscope/cache/native-local-project",
+      tempRoot: "/tmp/bandscope/temp/native-local-project",
+      source: {
+        sourcePath: "/tmp/bandscope/input.wav",
+        fileName: "input.wav",
+        extension: "wav",
+        fileSizeBytes: MAX_LOCAL_AUDIO_FILE_BYTES + 1
+      }
+    });
+
+    const selection = await selectLocalAudioSource();
+
+    expect(selection).toEqual({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        message: "Selected audio file exceeds the 100 MiB analysis limit."
+      }
+    });
+  });
+
+  it("rejects an oversized native YouTube import before it becomes project state", async () => {
+    tauriWindow.__TAURI_INVOKE__ = vi.fn().mockResolvedValue({
+      projectId: "native-youtube-project",
+      sourceMode: "reference",
+      projectRoot: "/tmp/bandscope/projects/native-youtube-project",
+      cacheRoot: "/tmp/bandscope/cache/native-youtube-project",
+      tempRoot: "/tmp/bandscope/temp/native-youtube-project",
+      source: {
+        sourcePath: "/tmp/bandscope/temp/native-youtube-project/youtube.wav",
+        fileName: "youtube.wav",
+        extension: "wav",
+        fileSizeBytes: MAX_LOCAL_AUDIO_FILE_BYTES + 1
+      }
+    });
+
+    const selection = await importYoutubeUrl("https://youtu.be/4ozX4yFUC34");
+
+    expect(selection).toEqual({
+      ok: false,
+      error: {
+        code: "invalid_request",
+        message: "Selected audio file exceeds the 100 MiB analysis limit."
+      }
+    });
+  });
+
   it("imports a standard YouTube URL through the browser fallback when Tauri is absent", async () => {
     const selection = await importYoutubeUrl("https://www.youtube.com/watch?v=4ozX4yFUC34");
 
diff --git a/apps/desktop/src/lib/analysis.ts b/apps/desktop/src/lib/analysis.ts
index bb750b34b..6ff443320 100644
--- a/apps/desktop/src/lib/analysis.ts
+++ b/apps/desktop/src/lib/analysis.ts
@@ -35,8 +35,14 @@ const BROWSER_PROGRESS_STEPS = [
   { progressLabel: "Saving reusable features", progressStage: "persist", progressPercent: 90 }
 ] as const;
 const UNSUPPORTED_LOCAL_AUDIO_MESSAGE = "Choose a WAV, MP3, FLAC, or M4A file to start analysis.";
+const LOCAL_AUDIO_TOO_LARGE_MESSAGE = "Selected audio file exceeds the 100 MiB analysis limit.";
+const LOCAL_AUDIO_POLICY_MESSAGE =
+  "Selected audio file metadata violates the analysis resource policy.";
+const MAX_LOCAL_AUDIO_FILE_BYTES = 100 * 1024 * 1024;
 const SAFE_LOCAL_AUDIO_MESSAGES = new Set([
   UNSUPPORTED_LOCAL_AUDIO_MESSAGE,
+  LOCAL_AUDIO_TOO_LARGE_MESSAGE,
+  LOCAL_AUDIO_POLICY_MESSAGE,
   "Could not read the selected audio file.",
   "Could not prepare the local project workspace.",
   "Could not prepare the local cache workspace.",
@@ -45,7 +51,7 @@ const SAFE_LOCAL_AUDIO_MESSAGES = new Set([
 const YOUTUBE_VIDEO_ID_PATTERN = /^[A-Za-z0-9_-]{11}$/;
 const MAX_YOUTUBE_URL_LENGTH = 2000;
 
-export { MAX_YOUTUBE_URL_LENGTH };
+export { MAX_LOCAL_AUDIO_FILE_BYTES, MAX_YOUTUBE_URL_LENGTH };
 
 /** Documented. */
 export type LocalAudioSelectionResult =
@@ -217,6 +223,26 @@ async function invokeAnalysis(command: string, args?: Record<string, unknown>):
   return browserFallback(command, args);
 }
 
+/**
+ * Parse a native/import bootstrap and enforce policy-v1 encoded-byte parity
+ * before the selection is allowed to become desktop project state.
+ *
+ * Python service and descriptor checks remain authoritative for analysis; this
+ * bridge check is defense in depth so local-file and imported-file intake fail
+ * at the same 100 MiB boundary instead of waiting for a later analysis stage.
+ */
+function parseBoundedAudioBootstrap(response: unknown): ProjectBootstrapSummary {
+  const bootstrap = parseProjectBootstrapSummary(response);
+  const fileSizeBytes = bootstrap.source.fileSizeBytes;
+  if (!Number.isSafeInteger(fileSizeBytes)) {
+    throw new Error(LOCAL_AUDIO_POLICY_MESSAGE);
+  }
+  if (fileSizeBytes > MAX_LOCAL_AUDIO_FILE_BYTES) {
+    throw new Error(LOCAL_AUDIO_TOO_LARGE_MESSAGE);
+  }
+  return bootstrap;
+}
+
 /** Documented. */
 export function createDefaultAnalysisRequest(): AnalysisJobRequest {
   return createDemoAnalysisJobRequest();
@@ -228,7 +254,7 @@ export async function selectLocalAudioSource(): Promise<LocalAudioSelectionResul
     const response = await invokeAnalysis("select_local_audio_source");
     return {
       ok: true,
-      bootstrap: parseProjectBootstrapSummary(response)
+      bootstrap: parseBoundedAudioBootstrap(response)
     };
   } catch (error) {
     return {
@@ -328,7 +354,7 @@ export async function importYoutubeUrl(url: string): Promise<LocalAudioSelection
     const response = await invokeAnalysis("import_youtube_url", { url });
     return {
       ok: true,
-      bootstrap: parseProjectBootstrapSummary(response)
+      bootstrap: parseBoundedAudioBootstrap(response)
     };
   } catch (error) {
     const message = error instanceof Error ? error.message : (typeof error === "string" ? error : "YouTube import failed.");
diff --git a/docs/doctoring/audio-resource-policy.md b/docs/doctoring/audio-resource-policy.md
new file mode 100644
index 000000000..f13dad669
--- /dev/null
+++ b/docs/doctoring/audio-resource-policy.md
@@ -0,0 +1,32 @@
+# Audio resource policy evidence
+
+## Scope
+
+This note records the external evidence used by BandScope's versioned local-audio resource policy. It is implementation doctoring, not a claim that BandScope is certified against any external standard.
+
+The current Python policy accepts at most 100 MiB of encoded local-audio input and at most 900 seconds of mono decoded audio at 44.1 kHz, which is 39,690,000 samples. Request metadata is an early rejection hint only: the decoder boundary still checks the opened descriptor's actual size. Decoder calls request one sample beyond the accepted duration and the returned waveform is then validated as a one-dimensional, non-empty, finite floating-point array at exactly 44.1 kHz and no longer than the accepted sample budget before beat tracking or Demucs inference. Policy construction also rejects byte, rate, duration, or derived sample-count limits that cannot be represented within the host's bounded integer/sample-count model, so extreme integer configuration cannot escape through Python-to-float conversion overflow. YouTube import uses the same encoded-byte ceiling: yt-dlp `max_filesize`, a progress hook that aborts once `downloaded_bytes` / `total_bytes` / `total_bytes_estimate` exceed 100 MiB, a pre-download reject on announced `filesize` / `filesize_approx`, and a post-download `AudioResourcePolicy` check that deletes the artifact. Native local-file and YouTube bootstrap then re-check the filesystem-observed length before storing project state.
+
+## Evidence-to-control mapping
+
+| Evidence | BandScope control |
+| --- | --- |
+| CWE-770 recommends explicit minimum/maximum resource expectations and limiting resources reachable by unprivileged actors. | `AudioResourcePolicy` makes encoded bytes, decoded samples, sample rate, numeric dtype, shape, finiteness, and checked limit arithmetic explicit fail-closed invariants. CWE-770 is the more specific mapping-friendly weakness beneath the broader CWE-400 resource-consumption class. |
+| OWASP ASVS v5.0.0-5.1.1 requires file-handling documentation to define accepted types/extensions and maximum size; v5.0.0-5.2.1 requires accepting only file sizes that can be processed without performance loss or denial of service. | BandScope documents and enforces a finite encoded-byte ceiling before decode, while retaining authoritative descriptor checks at the actual file boundary. ASVS targets web applications/services, so BandScope uses these requirements as security-engineering guidance rather than claiming ASVS conformance for the desktop product. |
+| librosa 0.11.0 documents `load(..., duration=...)` as loading only up to the requested duration and returning an ndarray plus the resulting sample rate. | Temporal analysis and stem separation request `max_duration + one sample` as a probe, then reject any returned waveform whose exact decoded sample count exceeds the accepted limit. The post-decode check remains authoritative because a duration argument alone is not treated as proof of resource-policy compliance. |
+| yt-dlp documents `max_filesize` as a FileDownloader parameter that skips files larger than the configured byte count, and `progress_hooks` as callbacks that receive `downloaded_bytes`, `total_bytes`, and `total_bytes_estimate`. | YouTube import sets `max_filesize` to `DEFAULT_MAX_ENCODED_FILE_BYTES`, aborts from the progress hook when those byte fields exceed the ceiling, and still revalidates the written file with the canonical policy so a missing or lying size announcement cannot bypass admission. |
+
+## Residual risk and follow-up
+
+This policy now bounds Python decode/model entry, native local-file bootstrap, and YouTube download/bootstrap encoded-byte admission. In-flight abort also deletes owned `.part`, `.ytdl`, and `-Frag*` siblings that stay inside that import's output directory; paths that escape the directory are ignored. Remaining #781 work is source channel/rate metadata contracts, decoded-memory estimates, CPU/GPU admission budgets, cancellation/resource measurements, and whole-product CPU/GPU parity evidence. Do not treat a post-download-only size check as sufficient: in-flight abort and owned-partial deletion must stay in place so an unknown-size transfer cannot fill the cache root.
+
+## References
+
+librosa development team. (2025). *librosa.load (librosa 0.11.0)* [Documentation]. https://librosa.org/doc/0.11.0/generated/librosa.load.html
+
+MITRE Corporation. (2026, April 30). *CWE-770: Allocation of resources without limits or throttling (Version 4.20).* Common Weakness Enumeration. https://cwe.mitre.org/data/definitions/770.html
+
+OWASP Foundation. (2025, May). *OWASP Application Security Verification Standard 5.0.0.* https://github.com/OWASP/ASVS/tree/v5.0.0_release/5.0
+
+yt-dlp contributors. (2026). *FileDownloader parameters (`max_filesize`)* [Source documentation]. https://github.com/yt-dlp/yt-dlp/blob/master/yt_dlp/downloader/common.py
+
+yt-dlp contributors. (2026). *YoutubeDL `progress_hooks`* [Source documentation]. https://github.com/yt-dlp/yt-dlp/blob/master/yt_dlp/YoutubeDL.py
diff --git a/docs/security/app-security.md b/docs/security/app-security.md
index a9983fb97..0bc942986 100644
--- a/docs/security/app-security.md
+++ b/docs/security/app-security.md
@@ -137,6 +137,8 @@ Every boundary crossing requires validation, scope restriction, minimal logging,
 - Cross-check extension, MIME, and actual decode behavior.
 - Prefer isolated worker processing for decode and analysis.
 - Guard against very large files, abnormal duration, and hostile metadata.
+- Apply the versioned canonical local-audio resource policy consistently at request preflight and again at the opened-file/decoded-waveform boundary; request metadata is never authoritative for actual resource use.
+- In the Python analysis boundary, reject decoded audio that is empty, non-finite, wrong-rate, wrong-shaped, or over the accepted sample budget before beat tracking or model inference. Use the one-sample-over decode probe described in `docs/doctoring/audio-resource-policy.md` so an exact-boundary track remains accepted while excess decoded output is observable and fails closed.
 - Do not add arbitrary filesystem scanning just to find media files.
 - When bootstrapping a project around local audio, prefer referencing the validated original file plus app-owned temp/cache/project roots over copying the file until persistence requirements justify the extra storage boundary.
 
@@ -147,6 +149,8 @@ Every boundary crossing requires validation, scope restriction, minimal logging,
 - Validate scheme, host, path, and query before any fetch or handoff.
 - Do not widen URL intake into a generic remote downloader.
 - Sanitize remote metadata before display.
+- Apply the same canonical 100 MiB encoded-byte ceiling during YouTube download as local-file intake. Abort with yt-dlp `max_filesize` and a progress hook, then delete owned `.part` / `.ytdl` / `-Frag*` siblings that stay inside that import directory. Do not keep a divergent post-download-only 50 MB limit that lets a large transfer fill the cache root first.
+- Revalidate the filesystem-observed downloaded length before storing bootstrap state. Treat announced `filesize` / `filesize_approx` as a pre-download hint only.
 
 ### Subprocesses and native tools
 
diff --git a/services/analysis-engine/src/bandscope_analysis/api.py b/services/analysis-engine/src/bandscope_analysis/api.py
index b376de293..217e27a42 100644
--- a/services/analysis-engine/src/bandscope_analysis/api.py
+++ b/services/analysis-engine/src/bandscope_analysis/api.py
@@ -14,6 +14,7 @@
 
 import numpy as np
 
+from bandscope_analysis.audio_resource_policy import DEFAULT_AUDIO_RESOURCE_POLICY
 from bandscope_analysis.health import HealthReport, build_health_report
 from bandscope_analysis.roles import RoleExtractor
 from bandscope_analysis.sections import extract_sections
@@ -306,8 +307,12 @@ def validate_analysis_job_request(payload: object) -> AnalysisJobRequest:
         raise ValueError("Invalid analysis job request: invalid field 'localSource.fileName'")
     if extension not in {"wav", "mp3", "flac", "m4a"}:
         raise ValueError("Invalid analysis job request: invalid field 'localSource.extension'")
-    if not isinstance(file_size_bytes, int) or file_size_bytes <= 0:
-        raise ValueError("Invalid analysis job request: invalid field 'localSource.fileSizeBytes'")
+    try:
+        file_size_bytes = DEFAULT_AUDIO_RESOURCE_POLICY.validate_encoded_file_bytes(file_size_bytes)
+    except ValueError as error:
+        raise ValueError(
+            "Invalid analysis job request: invalid field 'localSource.fileSizeBytes'"
+        ) from error
 
     normalized: AnalysisJobRequest = {
         "sourceKind": source_kind,
diff --git a/services/analysis-engine/src/bandscope_analysis/audio_resource_policy.py b/services/analysis-engine/src/bandscope_analysis/audio_resource_policy.py
new file mode 100644
index 000000000..1fae11c4f
--- /dev/null
+++ b/services/analysis-engine/src/bandscope_analysis/audio_resource_policy.py
@@ -0,0 +1,166 @@
+"""Canonical resource admission policy for local audio analysis.
+
+The policy is intentionally independent of individual analyzers. Expensive
+feature code consumes a decoded artifact only after encoded-file and decoded
+output checks agree on the same versioned limits. This prevents temporal,
+separation, chord, and register features from silently inventing incompatible
+resource ceilings.
+
+Security Notes:
+- Encoded byte counts are validated before decode/allocation work when the
+  opened file descriptor can provide an authoritative size.
+- Decoded audio is revalidated because container metadata and decoder behavior
+  are untrusted; accepted artifacts are finite, mono, floating-point, at the
+  configured sample rate, and within the configured decoded-sample budget.
+- Decoders receive a one-sample-over-budget probe duration so a longer source is
+  rejected instead of being silently truncated to the accepted duration.
+- Policy arithmetic rejects unrepresentable limits before float/sample-count
+  conversion so malformed configuration cannot escape the stable failure mode.
+- Validation errors are payload-free and never include source paths or audio
+  content.
+"""
+
+from __future__ import annotations
+
+import math
+import sys
+from dataclasses import dataclass
+from typing import Any, cast
+
+import numpy as np
+from numpy.typing import NDArray
+
+AUDIO_RESOURCE_POLICY_VERSION = "1"
+DEFAULT_TARGET_SAMPLE_RATE = 44_100
+DEFAULT_MAX_ENCODED_FILE_BYTES = 100 * 1024 * 1024
+DEFAULT_MAX_DURATION_SECONDS = 15 * 60
+_POLICY_ERROR = "Audio input violates the audio resource policy."
+
+
+@dataclass(frozen=True)
+class AudioResourcePolicy:
+    """Versioned limits applied before and after local audio decoding.
+
+    Args:
+        max_encoded_file_bytes: Maximum non-empty encoded source size.
+        target_sample_rate: Required sample rate of the canonical decoded mono
+            artifact.
+        max_duration_seconds: Maximum decoded duration represented as a sample
+            ceiling at ``target_sample_rate``.
+    """
+
+    max_encoded_file_bytes: int = DEFAULT_MAX_ENCODED_FILE_BYTES
+    target_sample_rate: int = DEFAULT_TARGET_SAMPLE_RATE
+    max_duration_seconds: float = float(DEFAULT_MAX_DURATION_SECONDS)
+
+    def __post_init__(self) -> None:
+        """Reject invalid policy configuration before it can weaken admission."""
+        if (
+            isinstance(self.max_encoded_file_bytes, bool)
+            or not isinstance(self.max_encoded_file_bytes, int)
+            or self.max_encoded_file_bytes <= 0
+            or self.max_encoded_file_bytes > sys.maxsize - 1
+        ):
+            raise ValueError(_POLICY_ERROR)
+        if (
+            isinstance(self.target_sample_rate, bool)
+            or not isinstance(self.target_sample_rate, int)
+            or self.target_sample_rate <= 0
+            or self.target_sample_rate > sys.maxsize - 1
+        ):
+            raise ValueError(_POLICY_ERROR)
+        if isinstance(self.max_duration_seconds, bool) or not isinstance(
+            self.max_duration_seconds, int | float
+        ):
+            raise ValueError(_POLICY_ERROR)
+        try:
+            duration_seconds = float(self.max_duration_seconds)
+        except (OverflowError, ValueError):
+            raise ValueError(_POLICY_ERROR) from None
+        if not math.isfinite(duration_seconds) or duration_seconds <= 0.0:
+            raise ValueError(_POLICY_ERROR)
+        decoded_samples = self.target_sample_rate * duration_seconds
+        if (
+            not math.isfinite(decoded_samples)
+            or decoded_samples < 1.0
+            or decoded_samples > sys.maxsize - 1
+        ):
+            raise ValueError(_POLICY_ERROR)
+
+    @property
+    def max_decoded_samples(self) -> int:
+        """Return the maximum mono sample count allowed after decoding."""
+        return int(self.target_sample_rate * float(self.max_duration_seconds))
+
+    @property
+    def decode_probe_duration_seconds(self) -> float:
+        """Return a bounded decoder duration that includes one rejection probe sample."""
+        return (self.max_decoded_samples + 1) / self.target_sample_rate
+
+    def validate_encoded_file_bytes(self, file_size: object) -> int:
+        """Validate an authoritative encoded file size before decoding.
+
+        Args:
+            file_size: Byte count obtained from the already-open source file.
+
+        Returns:
+            The validated integer byte count.
+
+        Raises:
+            ValueError: If the value is not a positive integer within policy.
+        """
+        if (
+            isinstance(file_size, bool)
+            or not isinstance(file_size, int)
+            or file_size <= 0
+            or file_size > self.max_encoded_file_bytes
+        ):
+            raise ValueError(_POLICY_ERROR)
+        return file_size
+
+    def validate_decoded_audio(
+        self,
+        audio: object,
+        sample_rate: object,
+    ) -> NDArray[np.floating[Any]]:
+        """Revalidate the canonical decoded artifact before feature analysis.
+
+        Args:
+            audio: Candidate mono NumPy array returned by the decoder.
+            sample_rate: Decoder-reported sample rate in Hz.
+
+        Returns:
+            The original validated NumPy floating-point array without copying it.
+
+        Raises:
+            ValueError: If dtype, shape, sample rate, sample count, or finiteness
+                does not satisfy this policy.
+        """
+        if (
+            not isinstance(audio, np.ndarray)
+            or audio.ndim != 1
+            or audio.size == 0
+            or not np.issubdtype(audio.dtype, np.floating)
+        ):
+            raise ValueError(_POLICY_ERROR)
+        if (
+            isinstance(sample_rate, bool)
+            or not isinstance(sample_rate, int)
+            or sample_rate != self.target_sample_rate
+        ):
+            raise ValueError(_POLICY_ERROR)
+        if audio.size > self.max_decoded_samples or not np.isfinite(audio).all():
+            raise ValueError(_POLICY_ERROR)
+        return cast(NDArray[np.floating[Any]], audio)
+
+
+DEFAULT_AUDIO_RESOURCE_POLICY = AudioResourcePolicy()
+
+__all__ = [
+    "AUDIO_RESOURCE_POLICY_VERSION",
+    "AudioResourcePolicy",
+    "DEFAULT_AUDIO_RESOURCE_POLICY",
+    "DEFAULT_MAX_DURATION_SECONDS",
+    "DEFAULT_MAX_ENCODED_FILE_BYTES",
+    "DEFAULT_TARGET_SAMPLE_RATE",
+]
diff --git a/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py b/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py
index c36e0f1fc..7b5268ed8 100644
--- a/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py
+++ b/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py
@@ -9,6 +9,9 @@
 Security Notes:
 - Treats the selected audio file as untrusted input: the path is normalized and
   verified to be a file, and a maximum byte size is enforced before decode.
+- Decoded audio is revalidated against the same versioned resource policy before
+  Demucs/model work so overlong, malformed, or non-finite decoder output fails
+  closed instead of being silently truncated or normalized.
 - Inference runs locally on CPU with no network access. The model weights are
   loaded from the local Demucs cache or a configured bundled path; offline
   weight bundling is tracked in the supplemental component inventory.
@@ -31,9 +34,12 @@
 import librosa
 import numpy as np
 
+from bandscope_analysis.audio_resource_policy import (
+    DEFAULT_MAX_DURATION_SECONDS,
+    AudioResourcePolicy,
+)
 from bandscope_analysis.temporal.analyzer import (
     KNOWN_LIBROSA_NUMBA_WARNING_FILTERS,
-    MAX_ANALYSIS_DURATION_SECONDS,
     MAX_AUDIO_FILE_BYTES,
     TARGET_SR,
 )
@@ -63,7 +69,7 @@ class AudioSeparationConfig:
 
     target_sample_rate: int = TARGET_SR
     max_file_bytes: int = MAX_AUDIO_FILE_BYTES
-    max_duration_seconds: float = float(MAX_ANALYSIS_DURATION_SECONDS)
+    max_duration_seconds: float = float(DEFAULT_MAX_DURATION_SECONDS)
     model_name: str = "htdemucs"
     device: str = "cpu"
     # Demucs splits long audio into overlapping segments internally, bounding
@@ -75,8 +81,13 @@ class AudioStemSeparator:
     """Split a selected local mix into canonical stems for downstream analysis."""
 
     def __init__(self, config: AudioSeparationConfig | None = None) -> None:
-        """Initialize the local stem separator (model is loaded lazily)."""
+        """Initialize the local stem separator and its canonical resource policy."""
         self.config = config or AudioSeparationConfig()
+        self.resource_policy = AudioResourcePolicy(
+            max_encoded_file_bytes=self.config.max_file_bytes,
+            target_sample_rate=self.config.target_sample_rate,
+            max_duration_seconds=self.config.max_duration_seconds,
+        )
         self._model: Any = None
 
     def separate(self, audio_path: str | Path) -> AudioSeparationResult:
@@ -190,15 +201,16 @@ def _resolve_audio_file(self, audio_path: str | Path) -> Path:
         return path
 
     def _load_audio(self, path: Path) -> tuple[AudioStemArray, int]:
-        """Load bounded mono audio without logging or exposing the full source path."""
+        """Load and revalidate bounded mono audio before model inference."""
         try:
             with path.open("rb") as fileobj:
                 file_size = os.fstat(fileobj.fileno()).st_size
-                if file_size > self.config.max_file_bytes:
-                    raise ValueError(
-                        "Audio file is too large for stem separation: "
-                        f"{file_size} bytes (max {self.config.max_file_bytes} bytes)"
-                    )
+                if file_size <= 0:
+                    raise ValueError(f"Stem separation decode failed for {path.name}")
+                try:
+                    self.resource_policy.validate_encoded_file_bytes(file_size)
+                except ValueError as error:
+                    raise ValueError("Audio file is too large for stem separation") from error
 
                 with warnings.catch_warnings():
                     warnings.filterwarnings(
@@ -214,16 +226,19 @@ def _load_audio(self, path: Path) -> tuple[AudioStemArray, int]:
                         )
                     y, sr = librosa.load(
                         fileobj,
-                        sr=self.config.target_sample_rate,
+                        sr=self.resource_policy.target_sample_rate,
                         mono=True,
-                        duration=self.config.max_duration_seconds,
+                        duration=self.resource_policy.decode_probe_duration_seconds,
                     )
         except ValueError:
             raise
         except Exception as error:
             raise ValueError(f"Stem separation decode failed for {path.name}") from error
 
-        return _as_float_array(y), int(sr)
+        if isinstance(y, np.ndarray) and y.size == 0:
+            raise ValueError(f"Stem separation decode failed for {path.name}")
+        validated_audio = self.resource_policy.validate_decoded_audio(y, sr)
+        return _as_float_array(validated_audio), int(sr)
 
     def _fit_length(self, audio: AudioStemArray, target_length: int) -> AudioStemArray:
         """Trim or pad a stem to match the source length exactly."""
diff --git a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
index 7fe5ae6f7..1584517bd 100644
--- a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
+++ b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
@@ -12,14 +12,23 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from bandscope_analysis.audio_resource_policy import (
+    DEFAULT_AUDIO_RESOURCE_POLICY,
+    DEFAULT_MAX_DURATION_SECONDS,
+    AudioResourcePolicy,
+)
+
 from .model import TemporalFeatures
 
 logger = logging.getLogger(__name__)
 
-# Standard sample rate for BandScope analysis
-TARGET_SR = 44100
-MAX_AUDIO_FILE_BYTES = 100 * 1024 * 1024  # 100 MiB
-MAX_ANALYSIS_DURATION_SECONDS = 15 * 60  # 15 minutes
+# Compatibility aliases retained for callers/tests while the canonical values
+# are owned by AudioResourcePolicy. The decode-duration alias intentionally
+# includes one rejection-probe sample so an overlong source is detected rather
+# than silently truncated at the accepted rehearsal duration.
+TARGET_SR = DEFAULT_AUDIO_RESOURCE_POLICY.target_sample_rate
+MAX_AUDIO_FILE_BYTES = DEFAULT_AUDIO_RESOURCE_POLICY.max_encoded_file_bytes
+MAX_ANALYSIS_DURATION_SECONDS = DEFAULT_AUDIO_RESOURCE_POLICY.decode_probe_duration_seconds
 KNOWN_LIBROSA_NUMBA_WARNING_FILTERS = (
     (DeprecationWarning, r".*pkg_resources is deprecated.*", r".*librosa.*"),
     (FutureWarning, r".*Numba.*", r".*numba.*"),
@@ -57,10 +66,25 @@ def _estimate_downbeats(
 
 
 class TemporalAnalyzer:
-    """Analyzes temporal features (BPM, beats) from audio files."""
+    """Analyze bounded temporal features (BPM and beat grids) from local audio."""
+
+    def __init__(self, resource_policy: AudioResourcePolicy | None = None) -> None:
+        """Create an analyzer bound to one canonical local-audio resource policy.
+
+        Args:
+            resource_policy: Explicit policy for tests or specialized callers.
+                The default preserves the public module-level byte ceiling while
+                taking sample-rate and accepted rehearsal duration from the
+                canonical policy layer.
+        """
+        self.resource_policy = resource_policy or AudioResourcePolicy(
+            max_encoded_file_bytes=MAX_AUDIO_FILE_BYTES,
+            target_sample_rate=TARGET_SR,
+            max_duration_seconds=DEFAULT_MAX_DURATION_SECONDS,
+        )
 
     def analyze(self, audio_path: str | Path) -> TemporalFeatures:
-        """Decode audio and extract temporal features.
+        """Decode bounded audio and extract temporal features.
 
         Args:
             audio_path: Path to the audio file.
@@ -78,11 +102,10 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures:
         try:
             with path.open("rb") as fileobj:
                 file_size = os.fstat(fileobj.fileno()).st_size
-                if file_size > MAX_AUDIO_FILE_BYTES:
-                    raise ValueError(
-                        f"Audio file is too large for temporal analysis: {file_size} bytes "
-                        f"(max {MAX_AUDIO_FILE_BYTES} bytes)"
-                    )
+                try:
+                    self.resource_policy.validate_encoded_file_bytes(file_size)
+                except ValueError as error:
+                    raise ValueError("Audio file is too large for temporal analysis") from error
 
                 with warnings.catch_warnings():
                     warnings.filterwarnings(
@@ -99,26 +122,25 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures:
                             message=message,
                             module=module,
                         )
-                    # Load audio, converting to mono and standardizing sample rate
+                    # Decode one sample beyond the accepted duration so longer
+                    # sources fail closed instead of becoming silently truncated.
                     y, sr = librosa.load(
                         fileobj,
-                        sr=TARGET_SR,
+                        sr=self.resource_policy.target_sample_rate,
                         mono=True,
-                        duration=MAX_ANALYSIS_DURATION_SECONDS,
+                        duration=self.resource_policy.decode_probe_duration_seconds,
                     )
 
-            # Ensure it's a 1D float array for librosa
+            # Preserve the established diagnostic for decoder contract violations
+            # before applying the canonical numeric policy.
             if not isinstance(y, np.ndarray):
                 raise ValueError("Expected numpy array from librosa.load")
 
-            y_array: NDArray[np.floating[Any]] = y
+            y_array = self.resource_policy.validate_decoded_audio(y, sr)
             duration = float(librosa.get_duration(y=y_array, sr=sr))
 
             logger.info("Extracting tempo and beat tracking...")
-            # Use librosa's robust beat tracker
             tempo, beat_frames = librosa.beat.beat_track(y=y_array, sr=sr)
-
-            # Convert frame indices to time (seconds)
             beat_times: NDArray[np.floating[Any]] = librosa.frames_to_time(beat_frames, sr=sr)
 
             # Place downbeats on the strongest-onset bar phase (looks at the audio,
diff --git a/services/analysis-engine/src/bandscope_analysis/youtube.py b/services/analysis-engine/src/bandscope_analysis/youtube.py
index c98f4e513..aaf8fff0a 100644
--- a/services/analysis-engine/src/bandscope_analysis/youtube.py
+++ b/services/analysis-engine/src/bandscope_analysis/youtube.py
@@ -1,11 +1,29 @@
-"""
-YouTube import capabilities for BandScope.
+"""YouTube import capabilities for BandScope.
 
 This module provides a safe wrapper around yt-dlp to download audio from YouTube.
+
+Security Notes:
+    - URL intake remains host/path/query allowlisted before any network work.
+    - Encoded-byte admission uses the same canonical 100 MiB policy as local
+      audio. yt-dlp ``max_filesize`` and a progress hook abort in-flight
+      transfers so a multi-gigabyte download cannot fill the cache root before
+      the post-download check runs.
+    - Announced ``filesize`` / ``filesize_approx`` values over the policy
+      ceiling reject the import before ``download=True``.
+    - The opened-file size is revalidated with ``AudioResourcePolicy`` after
+      download; oversize artifacts are deleted.
+    - In-flight abort deletes owned ``tmpfilename`` / ``filename`` siblings
+      (``.part``, ``.ytdl``, ``-Frag*``) that stay inside this import's
+      ``out_dir``. Paths that escape the directory are ignored.
+    - Validation errors are payload-free and never include source paths, URLs,
+      cookies, or audio content.
 """
 
+from __future__ import annotations
+
 import argparse
 import json
+import math
 import os
 import re
 import sys
@@ -14,6 +32,12 @@
 
 import yt_dlp  # type: ignore
 
+from bandscope_analysis.audio_resource_policy import (
+    DEFAULT_AUDIO_RESOURCE_POLICY,
+    DEFAULT_MAX_DURATION_SECONDS,
+    DEFAULT_MAX_ENCODED_FILE_BYTES,
+)
+
 YOUTUBE_VIDEO_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{11}$")
 MAX_YOUTUBE_URL_LENGTH = 2000
 SUPPORTED_AUDIO_EXTENSIONS = (".opus", ".m4a", ".mp3", ".wav", ".aac", ".flac", ".ogg")
@@ -21,6 +45,22 @@
     "Failed to download audio from YouTube. Please use a local audio file instead."
 )
 YOUTUBE_IMPORT_FAILED_MESSAGE = "YouTube import failed. Please use a local audio file instead."
+YOUTUBE_SIZE_EXCEEDED_MESSAGE = "Selected audio file exceeds the 100 MiB analysis limit."
+
+
+class YoutubeResourceLimitError(Exception):
+    """Fail-closed YouTube admission error that never includes payload paths."""
+
+    def __init__(self, code: str, message: str) -> None:
+        """Store a payload-safe public error code and next-action message.
+
+        Args:
+            code: Stable machine-readable error code.
+            message: User-facing instruction that omits paths and URLs.
+        """
+        super().__init__(message)
+        self.code = code
+        self.message = message
 
 
 def validate_url(url: str) -> bool:
@@ -72,9 +112,166 @@ def _find_downloaded_file(actual_filepath: str) -> Optional[str]:
     return actual_filepath
 
 
+def _size_exceeded_result() -> Dict[str, Any]:
+    """Return the payload-safe oversize result shared by every admission path."""
+    return {
+        "ok": False,
+        "error": {
+            "code": "size_exceeded",
+            "message": YOUTUBE_SIZE_EXCEEDED_MESSAGE,
+        },
+    }
+
+
+def _announced_size_exceeds_policy(announced: object) -> bool:
+    """Return whether yt-dlp metadata already reports an over-budget file.
+
+    Args:
+        announced: Candidate ``filesize`` or ``filesize_approx`` value.
+
+    Returns:
+        True when the value is a finite number strictly above the policy ceiling.
+    """
+    if isinstance(announced, bool) or not isinstance(announced, int | float):
+        return False
+    if isinstance(announced, float) and not math.isfinite(announced):
+        return False
+    size_bytes: int | float = announced
+    return bool(size_bytes > DEFAULT_MAX_ENCODED_FILE_BYTES)
+
+
+def _reject_announced_oversize(info: dict[str, Any]) -> Dict[str, Any] | None:
+    """Reject before download when extract_info already announced oversize bytes.
+
+    Args:
+        info: Metadata dictionary from ``extract_info(..., download=False)``.
+
+    Returns:
+        The size-exceeded result, or ``None`` when download may proceed.
+    """
+    if _announced_size_exceeds_policy(info.get("filesize")) or _announced_size_exceeds_policy(
+        info.get("filesize_approx")
+    ):
+        return _size_exceeded_result()
+    return None
+
+
+def _owned_file_path(path: object, out_dir: str) -> str | None:
+    """Return a real path only when it stays inside this import's output directory.
+
+    Args:
+        path: Candidate filesystem path from yt-dlp status or sibling lookup.
+        out_dir: Directory passed to this import call.
+
+    Returns:
+        The resolved file path, or ``None`` when the value is unsafe or foreign.
+    """
+    if not isinstance(path, str) or path == "":
+        return None
+    try:
+        resolved = os.path.realpath(path)
+        root = os.path.realpath(out_dir)
+    except OSError:
+        return None
+    if resolved == root or not resolved.startswith(root + os.sep):
+        return None
+    return resolved
+
+
+def _remove_owned_file(path: object, out_dir: str) -> None:
+    """Delete one owned regular file, ignoring missing-path races.
+
+    Args:
+        path: Candidate path that must resolve inside ``out_dir``.
+        out_dir: Directory passed to this import call.
+    """
+    owned = _owned_file_path(path, out_dir)
+    if owned is None:
+        return
+    try:
+        if os.path.isfile(owned):
+            os.remove(owned)
+    except OSError:
+        return
+
+
+def _remove_download_artifacts(status: dict[str, Any], out_dir: str) -> None:
+    """Delete the current download's partial, fragment, and control files.
+
+    Args:
+        status: yt-dlp progress-hook payload that may name ``tmpfilename``
+            and ``filename``.
+        out_dir: Directory passed to this import call.
+    """
+    stems: set[str] = set()
+    for key in ("tmpfilename", "filename"):
+        owned = _owned_file_path(status.get(key), out_dir)
+        if owned is None:
+            continue
+        _remove_owned_file(owned, out_dir)
+        name = os.path.basename(owned)
+        if name.endswith(".part"):
+            name = name[: -len(".part")]
+        stems.add(name)
+    if not stems:
+        return
+    try:
+        entries = os.listdir(out_dir)
+    except OSError:
+        return
+    for entry in entries:
+        matches_stem = any(
+            entry == stem or entry.startswith(f"{stem}.") or entry.startswith(f"{stem}-")
+            for stem in stems
+        )
+        if matches_stem:
+            _remove_owned_file(os.path.join(out_dir, entry), out_dir)
+
+
+def _abort_over_budget_download(status: dict[str, Any], out_dir: str) -> None:
+    """Abort an in-flight download once encoded bytes exceed the policy ceiling.
+
+    Args:
+        status: yt-dlp progress-hook payload. Unknown statuses are ignored.
+        out_dir: Directory passed to this import call, used to delete partials.
+    """
+    if status.get("status") not in {"downloading", "finished"}:
+        return
+    for key in ("downloaded_bytes", "total_bytes", "total_bytes_estimate"):
+        candidate = status.get(key)
+        if isinstance(candidate, bool) or not isinstance(candidate, int):
+            continue
+        if candidate > DEFAULT_MAX_ENCODED_FILE_BYTES:
+            _remove_download_artifacts(status, out_dir)
+            raise YoutubeResourceLimitError("size_exceeded", YOUTUBE_SIZE_EXCEEDED_MESSAGE)
+
+
+def _make_abort_hook(out_dir: str) -> Any:
+    """Bind the in-flight abort hook to one import output directory.
+
+    Args:
+        out_dir: Directory passed to this import call.
+
+    Returns:
+        A yt-dlp progress hook that aborts and deletes owned partials.
+    """
+
+    def _bound_abort_over_budget_download(status: dict[str, Any]) -> None:
+        """Abort and delete owned partials for this import directory.
+
+        Args:
+            status: yt-dlp progress-hook payload.
+        """
+        _abort_over_budget_download(status, out_dir)
+
+    return _bound_abort_over_budget_download
+
+
 def _handle_download_error(e: yt_dlp.utils.DownloadError) -> Dict[str, Any]:
     """Map yt-dlp DownloadError to the public YouTube import error response."""
     msg = str(e).lower()
+    if "max-filesize" in msg or "100 mib" in msg:
+        return _size_exceeded_result()
     if (
         "sign in" in msg
         or "members-only" in msg
@@ -130,6 +327,8 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
         "noplaylist": True,
         "postprocessors": [{"key": "FFmpegExtractAudio"}],
         "geo_bypass": False,
+        "max_filesize": DEFAULT_MAX_ENCODED_FILE_BYTES,
+        "progress_hooks": [_make_abort_hook(out_dir)],
     }
 
     try:
@@ -138,7 +337,7 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
             if info is None:
                 raise Exception("Failed to extract info")
             duration = info.get("duration")
-            if duration is not None and duration > 15 * 60:
+            if duration is not None and duration > DEFAULT_MAX_DURATION_SECONDS:
                 return {
                     "ok": False,
                     "error": {
@@ -146,6 +345,9 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
                         "message": "Video exceeds the 15-minute limit.",
                     },
                 }
+            announced_rejection = _reject_announced_oversize(info)
+            if announced_rejection is not None:
+                return announced_rejection
 
             info = ydl.extract_info(url, download=True)
             if info is None:
@@ -163,18 +365,14 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
                     },
                 }
 
-            if (
-                os.path.exists(actual_filepath)
-                and os.path.getsize(actual_filepath) > 50 * 1024 * 1024
-            ):
-                os.remove(actual_filepath)
-                return {
-                    "ok": False,
-                    "error": {
-                        "code": "size_exceeded",
-                        "message": "Downloaded file exceeds the 50MB limit.",
-                    },
-                }
+            try:
+                DEFAULT_AUDIO_RESOURCE_POLICY.validate_encoded_file_bytes(
+                    os.path.getsize(actual_filepath)
+                )
+            except ValueError:
+                if os.path.exists(actual_filepath):
+                    os.remove(actual_filepath)
+                return _size_exceeded_result()
             return {
                 "ok": True,
                 "metadata": {
@@ -184,6 +382,8 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
                     "filepath": actual_filepath,
                 },
             }
+    except YoutubeResourceLimitError:
+        return _size_exceeded_result()
     except yt_dlp.utils.DownloadError as e:
         return _handle_download_error(e)
     except Exception:
diff --git a/services/analysis-engine/tests/test_audio_resource_policy.py b/services/analysis-engine/tests/test_audio_resource_policy.py
new file mode 100644
index 000000000..75e2871df
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy.py
@@ -0,0 +1,101 @@
+"""Tests for the canonical local-audio resource policy."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from bandscope_analysis.audio_resource_policy import (
+    AUDIO_RESOURCE_POLICY_VERSION,
+    DEFAULT_AUDIO_RESOURCE_POLICY,
+    AudioResourcePolicy,
+)
+
+
+def test_default_policy_has_stable_version_and_rehearsal_budget() -> None:
+    """The default policy exposes one versioned budget shared by analyzers."""
+    assert AUDIO_RESOURCE_POLICY_VERSION == "1"
+    assert DEFAULT_AUDIO_RESOURCE_POLICY.max_encoded_file_bytes == 100 * 1024 * 1024
+    assert DEFAULT_AUDIO_RESOURCE_POLICY.target_sample_rate == 44_100
+    assert DEFAULT_AUDIO_RESOURCE_POLICY.max_duration_seconds == 15 * 60
+    assert DEFAULT_AUDIO_RESOURCE_POLICY.max_decoded_samples == 44_100 * 15 * 60
+
+
+@pytest.mark.parametrize("file_size", [True, -1, 0, 101])
+def test_encoded_file_size_fails_closed_outside_policy(file_size: object) -> None:
+    """Invalid, empty, or oversized encoded inputs are rejected before decode."""
+    policy = AudioResourcePolicy(max_encoded_file_bytes=100)
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        policy.validate_encoded_file_bytes(file_size)
+
+
+def test_encoded_file_size_accepts_exact_boundary() -> None:
+    """A non-empty encoded file exactly at the configured ceiling is accepted."""
+    policy = AudioResourcePolicy(max_encoded_file_bytes=100)
+
+    assert policy.validate_encoded_file_bytes(100) == 100
+
+
+@pytest.mark.parametrize(
+    ("audio", "sample_rate"),
+    [
+        (np.zeros(8_001, dtype=np.float32), 8_000),
+        (np.zeros((2, 4_000), dtype=np.float32), 8_000),
+        (np.array([0.0, np.nan], dtype=np.float32), 8_000),
+        (np.array(["not-a-sample"], dtype=object), 8_000),
+        (np.zeros(10, dtype=np.int16), 8_000),
+        (np.zeros(10, dtype=np.float32), 0),
+        (np.zeros(10, dtype=np.float32), True),
+    ],
+)
+def test_decoded_audio_fails_closed_outside_policy(
+    audio: np.ndarray,
+    sample_rate: object,
+) -> None:
+    """Decoded output is revalidated for type, shape, finiteness, rate, and sample budget."""
+    policy = AudioResourcePolicy(target_sample_rate=8_000, max_duration_seconds=1.0)
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        policy.validate_decoded_audio(audio, sample_rate)
+
+
+def test_decoded_audio_accepts_exact_sample_boundary() -> None:
+    """A finite mono artifact exactly at the decoded-sample ceiling is accepted."""
+    policy = AudioResourcePolicy(target_sample_rate=8_000, max_duration_seconds=1.0)
+    audio = np.zeros(8_000, dtype=np.float32)
+
+    validated = policy.validate_decoded_audio(audio, 8_000)
+
+    assert validated is audio
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"max_encoded_file_bytes": 0},
+        {"target_sample_rate": 0},
+        {"max_duration_seconds": 0.0},
+        {"max_duration_seconds": float("inf")},
+    ],
+)
+def test_policy_configuration_itself_fails_closed(kwargs: dict[str, object]) -> None:
+    """Invalid policy construction cannot silently create an unbounded budget."""
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioResourcePolicy(**kwargs)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"target_sample_rate": 10**400, "max_duration_seconds": 1.0},
+        {"target_sample_rate": 1, "max_duration_seconds": 10**400},
+        {"max_encoded_file_bytes": 10**400},
+    ],
+)
+def test_policy_configuration_fails_closed_on_unrepresentable_limits(
+    kwargs: dict[str, object],
+) -> None:
+    """Extreme integer limits cannot escape stable policy validation through overflow."""
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioResourcePolicy(**kwargs)  # type: ignore[arg-type]
diff --git a/services/analysis-engine/tests/test_audio_resource_policy_coverage_regressions.py b/services/analysis-engine/tests/test_audio_resource_policy_coverage_regressions.py
new file mode 100644
index 000000000..5bfbb0909
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy_coverage_regressions.py
@@ -0,0 +1,71 @@
+"""Coverage regressions for fail-closed audio resource admission branches."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from bandscope_analysis.audio_resource_policy import AudioResourcePolicy
+from bandscope_analysis.separation.audio_separator import (
+    AudioSeparationConfig,
+    AudioStemSeparator,
+)
+
+
+def test_policy_rejects_boolean_duration_configuration() -> None:
+    """A Boolean duration must not be coerced into a one-second resource budget."""
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioResourcePolicy(max_duration_seconds=True)
+
+
+def test_policy_rejects_less_than_one_decoded_sample_budget() -> None:
+    """A positive duration that represents less than one sample must fail closed."""
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioResourcePolicy(target_sample_rate=1, max_duration_seconds=0.5)
+
+
+def test_separator_rejects_empty_internal_loader_result_before_model(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An unexpected empty loader result must not reach Demucs inference."""
+    audio_path = tmp_path / "unexpected-empty.wav"
+    audio_path.write_bytes(b"not-empty")
+    separator = AudioStemSeparator(
+        AudioSeparationConfig(target_sample_rate=8_000, max_file_bytes=1_000_000)
+    )
+    monkeypatch.setattr(
+        separator,
+        "_load_audio",
+        lambda _path: (np.array([], dtype=np.float32), 8_000),
+    )
+
+    def fail_if_model_runs(_audio: np.ndarray, _sample_rate: int) -> dict[str, np.ndarray]:
+        raise AssertionError("empty decoded audio must be rejected before model inference")
+
+    monkeypatch.setattr(separator, "_separate_signal", fail_if_model_runs)
+
+    with pytest.raises(ValueError, match="Stem separation decode failed"):
+        separator.separate(audio_path)
+
+
+def test_separator_rejects_zero_byte_file_before_decoder(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A zero-byte selected source must fail before the decoder is invoked."""
+    audio_path = tmp_path / "empty.wav"
+    audio_path.write_bytes(b"")
+    separator = AudioStemSeparator(
+        AudioSeparationConfig(target_sample_rate=8_000, max_file_bytes=1_000_000)
+    )
+
+    def fail_if_decoder_runs(*_args: object, **_kwargs: object) -> tuple[np.ndarray, int]:
+        raise AssertionError("zero-byte input must be rejected before decoder invocation")
+
+    monkeypatch.setattr("bandscope_analysis.separation.audio_separator.librosa.load", fail_if_decoder_runs)
+
+    with pytest.raises(ValueError, match="Stem separation decode failed"):
+        separator.separate(audio_path)
diff --git a/services/analysis-engine/tests/test_audio_resource_policy_integration.py b/services/analysis-engine/tests/test_audio_resource_policy_integration.py
new file mode 100644
index 000000000..ec44a93d3
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy_integration.py
@@ -0,0 +1,163 @@
+"""Cross-boundary regressions for canonical local-audio resource admission."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from bandscope_analysis.api import validate_analysis_job_request
+from bandscope_analysis.audio_resource_policy import (
+    DEFAULT_AUDIO_RESOURCE_POLICY,
+    AudioResourcePolicy,
+)
+from bandscope_analysis.separation.audio_separator import (
+    AudioSeparationConfig,
+    AudioStemSeparator,
+)
+from bandscope_analysis.temporal.analyzer import TemporalAnalyzer
+
+
+def _local_request(file_size_bytes: object) -> dict[str, object]:
+    """Build one local-audio request whose only variable is encoded byte metadata."""
+    return {
+        "sourceKind": "local_audio",
+        "projectId": "policy-project",
+        "sourceLabel": "rehearsal.wav",
+        "roleFocus": [],
+        "localSource": {
+            "sourcePath": "/tmp/rehearsal.wav",
+            "fileName": "rehearsal.wav",
+            "extension": "wav",
+            "fileSizeBytes": file_size_bytes,
+        },
+    }
+
+
+@pytest.mark.parametrize(
+    "file_size_bytes",
+    [True, DEFAULT_AUDIO_RESOURCE_POLICY.max_encoded_file_bytes + 1],
+)
+def test_request_preflight_rejects_metadata_outside_canonical_policy(
+    file_size_bytes: object,
+) -> None:
+    """Reject impossible/oversized metadata before orchestration starts expensive work."""
+    with pytest.raises(ValueError, match="localSource.fileSizeBytes"):
+        validate_analysis_job_request(_local_request(file_size_bytes))
+
+
+def test_request_preflight_accepts_exact_encoded_byte_boundary() -> None:
+    """The service API accepts the same exact encoded-byte ceiling as the policy."""
+    request = validate_analysis_job_request(
+        _local_request(DEFAULT_AUDIO_RESOURCE_POLICY.max_encoded_file_bytes)
+    )
+
+    assert (
+        request["localSource"]["fileSizeBytes"]
+        == DEFAULT_AUDIO_RESOURCE_POLICY.max_encoded_file_bytes
+    )
+
+
+def test_temporal_decoder_probes_one_sample_past_duration_limit_and_fails_closed(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Temporal decode detects a one-sample-overlong source instead of silently truncating it."""
+    import librosa
+
+    policy = AudioResourcePolicy(
+        max_encoded_file_bytes=100,
+        target_sample_rate=8,
+        max_duration_seconds=1.0,
+    )
+    source = tmp_path / "overlong.wav"
+    source.write_bytes(b"bounded")
+    captured: dict[str, object] = {}
+
+    def fake_load(fileobj: object, **kwargs: object) -> tuple[np.ndarray, int]:
+        captured.update(kwargs)
+        return np.zeros(policy.max_decoded_samples + 1, dtype=np.float32), policy.target_sample_rate
+
+    monkeypatch.setattr(librosa, "load", fake_load)
+    monkeypatch.setattr(
+        librosa.beat,
+        "beat_track",
+        lambda **_: (_ for _ in ()).throw(
+            AssertionError("analysis must not run after policy rejection")
+        ),
+    )
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        TemporalAnalyzer(resource_policy=policy).analyze(source)
+
+    assert captured["duration"] == pytest.approx(
+        (policy.max_decoded_samples + 1) / policy.target_sample_rate
+    )
+    assert captured["sr"] == policy.target_sample_rate
+    assert captured["mono"] is True
+
+
+def test_stem_decoder_probes_one_sample_past_duration_limit_and_fails_closed(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Stem separation consumes the same decoded-sample ceiling as temporal analysis."""
+    import librosa
+
+    config = AudioSeparationConfig(
+        target_sample_rate=8,
+        max_file_bytes=100,
+        max_duration_seconds=1.0,
+    )
+    source = tmp_path / "overlong.wav"
+    source.write_bytes(b"bounded")
+    captured: dict[str, object] = {}
+
+    def fake_load(fileobj: object, **kwargs: object) -> tuple[np.ndarray, int]:
+        captured.update(kwargs)
+        return np.zeros(9, dtype=np.float32), 8
+
+    monkeypatch.setattr(librosa, "load", fake_load)
+    monkeypatch.setattr(
+        AudioStemSeparator,
+        "_separate_signal",
+        lambda *_: (_ for _ in ()).throw(
+            AssertionError("model must not run after policy rejection")
+        ),
+    )
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioStemSeparator(config).separate(source)
+
+    assert captured["duration"] == pytest.approx(9 / 8)
+    assert captured["sr"] == 8
+    assert captured["mono"] is True
+
+
+def test_stem_decoder_rejects_nonfinite_decoded_output_before_model(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Decoder NaN/Inf values fail closed instead of being normalized into model input."""
+    import librosa
+
+    source = tmp_path / "nonfinite.wav"
+    source.write_bytes(b"bounded")
+    monkeypatch.setattr(
+        librosa,
+        "load",
+        lambda *args, **kwargs: (np.array([0.0, np.nan], dtype=np.float32), 8),
+    )
+    monkeypatch.setattr(
+        AudioStemSeparator,
+        "_separate_signal",
+        lambda *_: (_ for _ in ()).throw(AssertionError("model must not receive non-finite audio")),
+    )
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        AudioStemSeparator(
+            AudioSeparationConfig(
+                target_sample_rate=8,
+                max_file_bytes=100,
+                max_duration_seconds=1.0,
+            )
+        ).separate(source)
diff --git a/services/analysis-engine/tests/test_youtube.py b/services/analysis-engine/tests/test_youtube.py
index 5531ac9d5..35b0d6689 100644
--- a/services/analysis-engine/tests/test_youtube.py
+++ b/services/analysis-engine/tests/test_youtube.py
@@ -2,12 +2,22 @@
 
 import importlib
 import sys
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 import yt_dlp  # type: ignore
 
-from bandscope_analysis.youtube import MAX_YOUTUBE_URL_LENGTH, download_youtube_audio, validate_url
+from bandscope_analysis.audio_resource_policy import DEFAULT_MAX_ENCODED_FILE_BYTES
+from bandscope_analysis.youtube import (
+    MAX_YOUTUBE_URL_LENGTH,
+    YOUTUBE_SIZE_EXCEEDED_MESSAGE,
+    _owned_file_path,
+    _remove_download_artifacts,
+    _remove_owned_file,
+    download_youtube_audio,
+    validate_url,
+)
 
 
 def test_validate_url() -> None:
@@ -89,6 +99,8 @@ def test_download_youtube_audio_success(
         "id": "abc123DEF45",
         "title": "Test Video",
         "duration": 60,
+        "filesize": True,
+        "filesize_approx": float("nan"),
     }
     mock_ydl.extract_info.return_value = mock_info
     mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.webm"
@@ -114,6 +126,8 @@ def test_download_youtube_audio_success(
     assert called_opts["noplaylist"] is True
     assert called_opts["geo_bypass"] is False
     assert called_opts["postprocessors"] == [{"key": "FFmpegExtractAudio"}]
+    assert called_opts["max_filesize"] == DEFAULT_MAX_ENCODED_FILE_BYTES
+    assert called_opts["progress_hooks"]
     assert "%(id)s.%(ext)s" in called_opts["outtmpl"]
 
     # Verify extract_info was called twice correctly: once for metadata, once for download
@@ -273,6 +287,49 @@ def test_download_youtube_audio_duration_exceeded(mock_ydl_class: MagicMock) ->
     assert result["error"]["code"] == "duration_exceeded"
 
 
+@patch("bandscope_analysis.youtube.os.path.getsize")
+@patch("bandscope_analysis.youtube.os.path.exists")
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_accepts_size_between_legacy_and_canonical_ceiling(
+    mock_ydl_class: MagicMock,
+    mock_exists: MagicMock,
+    mock_getsize: MagicMock,
+) -> None:
+    """A 60 MiB download that the old 50 MB check rejected is now accepted."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
+    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_getsize.return_value = 60 * 1024 * 1024
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is True
+    assert result["metadata"]["filepath"] == "/tmp/abc123DEF45.m4a"
+
+
+@patch("bandscope_analysis.youtube.os.path.getsize")
+@patch("bandscope_analysis.youtube.os.path.exists")
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_accepts_exact_policy_ceiling(
+    mock_ydl_class: MagicMock,
+    mock_exists: MagicMock,
+    mock_getsize: MagicMock,
+) -> None:
+    """An encoded YouTube file exactly at the 100 MiB ceiling is accepted."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
+    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_getsize.return_value = DEFAULT_MAX_ENCODED_FILE_BYTES
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is True
+
+
 @patch("bandscope_analysis.youtube.os.path.getsize")
 @patch("bandscope_analysis.youtube.os.path.exists")
 @patch("bandscope_analysis.youtube.os.remove")
@@ -283,20 +340,316 @@ def test_download_youtube_audio_size_exceeded(
     mock_exists: MagicMock,
     mock_getsize: MagicMock,
 ) -> None:
-    """Test download fails if size exceeds 50MB."""
+    """Post-download files one byte over the canonical 100 MiB ceiling are deleted."""
     mock_ydl = MagicMock()
     mock_ydl_class.return_value.__enter__.return_value = mock_ydl
     mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
     mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
     mock_exists.return_value = True
-    mock_getsize.return_value = 51 * 1024 * 1024
+    mock_getsize.return_value = DEFAULT_MAX_ENCODED_FILE_BYTES + 1
 
     result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
     assert result["ok"] is False
     assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
     mock_remove.assert_called_with("/tmp/abc123DEF45.m4a")
 
 
+@patch("bandscope_analysis.youtube.os.path.getsize")
+@patch("bandscope_analysis.youtube.os.path.exists")
+@patch("bandscope_analysis.youtube.os.remove")
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_oversize_skips_remove_when_file_already_gone(
+    mock_ydl_class: MagicMock,
+    mock_remove: MagicMock,
+    mock_exists: MagicMock,
+    mock_getsize: MagicMock,
+) -> None:
+    """A vanished oversize artifact still fails closed without a remove race."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
+    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
+    mock_exists.side_effect = [True, False]
+    mock_getsize.return_value = DEFAULT_MAX_ENCODED_FILE_BYTES + 1
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    mock_remove.assert_not_called()
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_rejects_announced_filesize_before_download(
+    mock_ydl_class: MagicMock,
+) -> None:
+    """Announced filesize over the policy ceiling must not start the download."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {
+        "id": "abc123DEF45",
+        "duration": 60,
+        "filesize": DEFAULT_MAX_ENCODED_FILE_BYTES + 1,
+    }
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+    mock_ydl.extract_info.assert_called_once_with(
+        "https://youtube.com/watch?v=abc123DEF45",
+        download=False,
+    )
+
+
+@pytest.mark.parametrize(
+    "info",
+    [
+        {
+            "id": "abc123DEF45",
+            "duration": 60,
+            "filesize_approx": DEFAULT_MAX_ENCODED_FILE_BYTES + 1,
+        },
+        {
+            "id": "abc123DEF45",
+            "duration": 60,
+            "filesize_approx": float(DEFAULT_MAX_ENCODED_FILE_BYTES) + 0.5,
+        },
+    ],
+)
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_rejects_announced_approximate_oversize(
+    mock_ydl_class: MagicMock,
+    info: dict[str, object],
+) -> None:
+    """Approximate oversize metadata rejects the import before download starts."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = info
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    mock_ydl.extract_info.assert_called_once()
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_progress_hook_aborts_over_budget(
+    mock_ydl_class: MagicMock,
+) -> None:
+    """In-flight progress that crosses the encoded-byte ceiling fails closed."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+
+    def extract_info(_url: str, download: bool = False) -> dict[str, object]:
+        """Invoke the registered progress hook when the download starts."""
+        if download:
+            hook = mock_ydl_class.call_args[0][0]["progress_hooks"][0]
+            hook(
+                {
+                    "status": "downloading",
+                    "downloaded_bytes": DEFAULT_MAX_ENCODED_FILE_BYTES + 1,
+                }
+            )
+        return {"id": "abc123DEF45", "duration": 60}
+
+    mock_ydl.extract_info.side_effect = extract_info
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_progress_hook_deletes_partial_artifacts(
+    mock_ydl_class: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """In-flight abort must delete written partials so they cannot fill the cache."""
+    out_dir = tmp_path / "import-cache"
+    out_dir.mkdir()
+    outsider = tmp_path / "unrelated-youtube-partial.part"
+    partial = out_dir / "abc123DEF45.m4a.part"
+    fragment = out_dir / "abc123DEF45.m4a-Frag1"
+    control = out_dir / "abc123DEF45.m4a.ytdl"
+    keep = out_dir / "keep-me.txt"
+    partial.write_bytes(b"partial-cache-bytes")
+    fragment.write_bytes(b"hls-fragment-bytes")
+    control.write_bytes(b"ytdl-control-bytes")
+    keep.write_bytes(b"unrelated-cache-note")
+    outsider.write_bytes(b"must-not-delete")
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+
+    def extract_info(_url: str, download: bool = False) -> dict[str, object]:
+        """Abort after yt-dlp has already written the current block to disk."""
+        if download:
+            hook = mock_ydl_class.call_args[0][0]["progress_hooks"][0]
+            hook(
+                {
+                    "status": "downloading",
+                    "downloaded_bytes": DEFAULT_MAX_ENCODED_FILE_BYTES + 1,
+                    "tmpfilename": str(partial),
+                    "filename": str(out_dir / "abc123DEF45.m4a"),
+                }
+            )
+        return {"id": "abc123DEF45", "duration": 60}
+
+    mock_ydl.extract_info.side_effect = extract_info
+
+    result = download_youtube_audio(
+        "https://youtube.com/watch?v=abc123DEF45",
+        str(out_dir),
+    )
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+    assert not partial.exists()
+    assert not fragment.exists()
+    assert not control.exists()
+    assert keep.exists()
+    assert outsider.exists()
+
+
+def test_owned_file_path_rejects_empty_foreign_and_unresolvable_paths(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Abort cleanup must not follow empty, escaped, or unresolvable paths."""
+    out_dir = tmp_path / "import-cache"
+    out_dir.mkdir()
+    escaped = tmp_path / "outside.part"
+    escaped.write_bytes(b"keep")
+
+    assert _owned_file_path(None, str(out_dir)) is None
+    assert _owned_file_path("", str(out_dir)) is None
+    assert _owned_file_path(str(out_dir), str(out_dir)) is None
+    assert _owned_file_path(str(escaped), str(out_dir)) is None
+
+    def boom(_path: str) -> str:
+        """Simulate a filesystem error while resolving a candidate path."""
+        raise OSError("realpath failed")
+
+    monkeypatch.setattr("bandscope_analysis.youtube.os.path.realpath", boom)
+    assert _owned_file_path(str(out_dir / "clip.part"), str(out_dir)) is None
+
+
+def test_remove_owned_file_ignores_missing_directories_and_remove_races(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Owned cleanup skips non-files and treats remove races as already gone."""
+    out_dir = tmp_path / "import-cache"
+    nested = out_dir / "nested-dir"
+    nested.mkdir(parents=True)
+    _remove_owned_file(None, str(out_dir))
+    _remove_owned_file(str(nested), str(out_dir))
+    assert nested.is_dir()
+
+    target = out_dir / "clip.part"
+    target.write_bytes(b"partial")
+
+    def boom(_path: str) -> None:
+        """Simulate a disappearing file during abort cleanup."""
+        raise OSError("remove failed")
+
+    monkeypatch.setattr("bandscope_analysis.youtube.os.remove", boom)
+    _remove_owned_file(str(target), str(out_dir))
+    assert target.exists()
+
+
+def test_remove_download_artifacts_skips_empty_status_and_unlistable_cache(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Artifact sweep no-ops when yt-dlp omitted paths or the cache vanished."""
+    out_dir = tmp_path / "import-cache"
+    out_dir.mkdir()
+    leftover = out_dir / "other-file.txt"
+    leftover.write_bytes(b"keep")
+    _remove_download_artifacts({"tmpfilename": None, "filename": 12}, str(out_dir))
+    assert leftover.exists()
+
+    partial = out_dir / "abc123DEF45.m4a.part"
+    partial.write_bytes(b"partial")
+
+    def boom(_path: str) -> list[str]:
+        """Simulate the import cache disappearing after the first delete."""
+        raise OSError("listdir failed")
+
+    monkeypatch.setattr("bandscope_analysis.youtube.os.listdir", boom)
+    _remove_download_artifacts({"tmpfilename": str(partial)}, str(out_dir))
+    assert leftover.exists()
+
+
+@patch("bandscope_analysis.youtube.os.path.getsize")
+@patch("bandscope_analysis.youtube.os.path.exists")
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_progress_hook_ignores_non_budget_updates(
+    mock_ydl_class: MagicMock,
+    mock_exists: MagicMock,
+    mock_getsize: MagicMock,
+) -> None:
+    """Unknown statuses and non-integer byte fields do not abort a valid download."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 60}
+    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_getsize.return_value = 10 * 1024 * 1024
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+    hook = mock_ydl_class.call_args[0][0]["progress_hooks"][0]
+    hook({"status": "error"})
+    hook({"status": "downloading", "downloaded_bytes": True})
+    hook({"status": "downloading", "downloaded_bytes": 12.5})
+    hook({"status": "downloading", "downloaded_bytes": 10})
+    hook({"status": "finished", "total_bytes": DEFAULT_MAX_ENCODED_FILE_BYTES})
+
+    assert result["ok"] is True
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_maps_max_filesize_download_error(
+    mock_ydl_class: MagicMock,
+) -> None:
+    """yt-dlp max-filesize aborts become the payload-safe size-exceeded result."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.side_effect = yt_dlp.utils.DownloadError(
+        "File is larger than max-filesize"
+    )
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+    assert "max-filesize" not in result["error"]["message"]
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_youtube_audio_maps_mib_limit_download_error(
+    mock_ydl_class: MagicMock,
+) -> None:
+    """Download errors that mention the 100 MiB ceiling stay payload-safe."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.side_effect = yt_dlp.utils.DownloadError(YOUTUBE_SIZE_EXCEEDED_MESSAGE)
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+
+
 def test_main_block(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
     """Test the CLI entry point."""
     test_args = [