diff --git a/CHANGELOG.md b/CHANGELOG.md
index eea696893..d45201760 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,15 @@
 - 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.
+- Preflight source-container duration, sample rate, and channel count from the already-open audio handle before temporal, stem, or bass-transcription decoders resample, downmix, or truncate it; successful metadata probes rewind the handle and malformed probes fail closed.
+- Bound the admitted canonical decoded mono buffer to 317,520,000 bytes as well as the existing 39,690,000-sample ceiling, so decoder dtype expansion cannot stay within the sample count while exceeding the explicit in-memory audio budget.
+- Fail closed on malformed known YouTube duration metadata before `download=True`; Boolean, non-numeric, non-finite, zero, negative, and non-canonical numeric-subtype duration evidence can no longer authorize a media download through Python numeric coercion or subclass semantics.
+- 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, reject a completed path that resolves outside the current import cache before post-download validation, cleanup, or success, and delete owned 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/architecture/overview.md b/docs/architecture/overview.md
index 3cf5261b9..b342805a5 100644
--- a/docs/architecture/overview.md
+++ b/docs/architecture/overview.md
@@ -37,6 +37,7 @@ GitHub is the source of truth for repository governance, PR review, CI/CD, Code
 - treat files, URLs, models, caches, and release artifacts as untrusted inputs
 - route orchestration through typed Tauri IPC and a narrow Python subprocess bridge before considering any loopback HTTP surface
 - bootstrap local audio projects by validating the selected file in Rust, then passing only typed source metadata through the orchestration boundary
+- before Python decoders transform source audio, preflight the already-open container handle through the shared `audio_resource_policy` source-rate/channel/duration contract, then rewind it for decoding
 - keep project and temp/cache bootstrap roots under Tauri-resolved app-owned directories rather than the shared OS temp namespace
 
 ## CI/CD and release flow
diff --git a/docs/doctoring/audio-resource-policy.md b/docs/doctoring/audio-resource-policy.md
new file mode 100644
index 000000000..f87d98fe0
--- /dev/null
+++ b/docs/doctoring/audio-resource-policy.md
@@ -0,0 +1,35 @@
+# 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. The canonical decoded NumPy buffer is additionally bounded to 317,520,000 bytes (39,690,000 samples × 8 bytes), so a decoder cannot stay under the sample ceiling while expanding the admitted in-memory artifact beyond the policy's explicit mono-buffer budget. Request metadata is an early rejection hint only: the decoder boundary still checks the opened descriptor's actual size. Before any `librosa.load(..., sr=..., mono=True, duration=...)` transformation, `soundfile.info` inspects the already-open source handle and the canonical policy rejects malformed headers, source rates below 8 kHz or above 192 kHz, source channel counts outside mono/stereo, and source duration beyond the path's limit; a successful probe rewinds the same handle. 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 the target rate and within both the accepted sample count and decoded-buffer byte budget before beat tracking, transcription, or Demucs inference. Policy construction also rejects byte, memory, 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. Before post-download duration/size checks, cleanup, or success metadata can use the yt-dlp result, the completed path is canonicalized and required to remain strictly beneath the current import `out_dir`; a foreign or escaped path fails closed without being deleted. 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, decoded mono-buffer bytes, 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 or in-memory byte size exceeds the accepted limits. The post-decode check remains authoritative because a duration argument alone is not treated as proof of resource-policy compliance. |
+| python-soundfile 0.13.1 documents `soundfile.info(file)` as returning container information, including sample rate, channels, duration, and frame count, without reading the decoded waveform. | `audio_metadata.preflight_audio_metadata` uses the already-open handle for source metadata admission, applies the shared rate/channel/duration policy, and rewinds the handle before `librosa.load`; parser and rewind failures become the canonical payload-free policy error. |
+| 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, validates that the completed path remains inside the per-import output directory, 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 source-container admission, decode/model entry by decoded sample count and decoded mono-buffer memory, and native local-file bootstrap plus 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. The completed-path containment check is a point-in-time canonical path check and does not claim descriptor/handle-level race freedom if a privileged local actor replaces filesystem entries after validation. The decoded-memory limit covers the admitted canonical NumPy audio artifact only; it does not claim to bound downstream temporary arrays, PyTorch tensors, model weights, or accelerator allocations. Remaining #781 work is explicit per-job CPU/GPU/VRAM 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
+
+python-soundfile contributors. (2025). *python-soundfile 0.13.1: `soundfile.info`* [Documentation]. https://python-soundfile.readthedocs.io/en/latest/
+
+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..bd50f0a00 100644
--- a/docs/security/app-security.md
+++ b/docs/security/app-security.md
@@ -137,6 +137,9 @@ 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.
+- Before any decoder resamples, downmixes, or duration-truncates local audio, inspect source-container metadata from the already-open handle with `soundfile.info`, enforce the shared 8 kHz–192 kHz and mono/stereo source contract, reject overlong sources, and rewind the handle before `librosa.load`.
+- 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 +150,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/__init__.py b/services/analysis-engine/src/bandscope_analysis/__init__.py
index 3867248e8..ce4beb801 100644
--- a/services/analysis-engine/src/bandscope_analysis/__init__.py
+++ b/services/analysis-engine/src/bandscope_analysis/__init__.py
@@ -1,6 +1,36 @@
 """BandScope analysis engine package."""
 
-from .api import get_analysis_status
+import logging
+from importlib import import_module
+
 from .health import build_health_report
 
+_STEM_SAFE_FAILURE_LOG_MESSAGES = frozenset(
+    {
+        "Stem separation failed because the source file was missing.",
+        "Stem separation unavailable because Demucs or torch is not installed.",
+        "Stem separation rejected invalid audio source data.",
+        "Stem separation failed with a runtime error.",
+        "Stem separation failed unexpectedly.",
+        "Stem separation failed before analysis job completion.",
+    }
+)
+
+
+class _ApiDiagnosticPrivacyFilter(logging.Filter):
+    """Redact traceback payloads only for known stem safe-failure diagnostics."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Preserve unrelated diagnostics while redacting owned safe-failure tracebacks."""
+        if record.getMessage() in _STEM_SAFE_FAILURE_LOG_MESSAGES:
+            record.exc_info = None
+            record.exc_text = None
+        return True
+
+
+_api_logger = logging.getLogger("bandscope_analysis.api")
+_api_logger.addFilter(_ApiDiagnosticPrivacyFilter())
+_api_module = import_module(".api", __name__)
+get_analysis_status = _api_module.get_analysis_status
+
 __all__ = ["build_health_report", "get_analysis_status"]
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_metadata.py b/services/analysis-engine/src/bandscope_analysis/audio_metadata.py
new file mode 100644
index 000000000..9f5874d8e
--- /dev/null
+++ b/services/analysis-engine/src/bandscope_analysis/audio_metadata.py
@@ -0,0 +1,44 @@
+"""Bounded source-container metadata preflight for local audio decoders.
+
+Security Notes:
+- The selected audio bytes and container headers are untrusted.
+- This module reads metadata from an already-open caller-owned handle only; it
+  does not open paths, decode PCM, follow URLs, or allocate a waveform.
+- Malformed headers, unsupported source rates/channels, and overlong sources
+  fail closed with the payload-free canonical policy error.
+- A successful probe rewinds the handle so the downstream decoder receives the
+  same source from its beginning.
+"""
+
+from __future__ import annotations
+
+from typing import BinaryIO
+
+import soundfile  # type: ignore[import-untyped]  # soundfile has no py.typed marker.
+
+from bandscope_analysis.audio_resource_policy import (
+    DEFAULT_AUDIO_RESOURCE_POLICY,
+    AudioResourcePolicy,
+)
+
+_POLICY_ERROR = "Audio input violates the audio resource policy."
+
+
+def preflight_audio_metadata(
+    fileobj: BinaryIO,
+    policy: AudioResourcePolicy = DEFAULT_AUDIO_RESOURCE_POLICY,
+) -> None:
+    """Validate source metadata without decoding PCM and rewind the handle."""
+    try:
+        fileobj.seek(0)
+        info = soundfile.info(fileobj)
+        fileobj.seek(0)
+        policy.validate_source_metadata(
+            frames=info.frames,
+            sample_rate=info.samplerate,
+            channels=info.channels,
+        )
+    except ValueError:
+        raise
+    except Exception as error:
+        raise ValueError(_POLICY_ERROR) from error
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..63e7194b6
--- /dev/null
+++ b/services/analysis-engine/src/bandscope_analysis/audio_resource_policy.py
@@ -0,0 +1,259 @@
+"""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 configured sample and memory budgets.
+- 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_MIN_SOURCE_SAMPLE_RATE = 8_000
+DEFAULT_MAX_SOURCE_SAMPLE_RATE = 192_000
+DEFAULT_MIN_SOURCE_CHANNELS = 1
+DEFAULT_MAX_SOURCE_CHANNELS = 2
+DEFAULT_MAX_ENCODED_FILE_BYTES = 100 * 1024 * 1024
+DEFAULT_MAX_DURATION_SECONDS = 15 * 60
+DEFAULT_MAX_DECODED_AUDIO_BYTES = (
+    DEFAULT_TARGET_SAMPLE_RATE * DEFAULT_MAX_DURATION_SECONDS * np.dtype(np.float64).itemsize
+)
+_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_decoded_audio_bytes: Maximum in-memory byte size of the canonical
+            decoded mono NumPy buffer.
+        min_source_sample_rate: Minimum source-container sample rate accepted
+            before resampling.
+        max_source_sample_rate: Maximum source-container sample rate accepted
+            before resampling.
+        min_source_channels: Minimum source-container channel count accepted
+            before downmixing.
+        max_source_channels: Maximum source-container channel count accepted
+            before downmixing.
+    """
+
+    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)
+    max_decoded_audio_bytes: int = DEFAULT_MAX_DECODED_AUDIO_BYTES
+    min_source_sample_rate: int = DEFAULT_MIN_SOURCE_SAMPLE_RATE
+    max_source_sample_rate: int = DEFAULT_MAX_SOURCE_SAMPLE_RATE
+    min_source_channels: int = DEFAULT_MIN_SOURCE_CHANNELS
+    max_source_channels: int = DEFAULT_MAX_SOURCE_CHANNELS
+
+    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)
+        if (
+            isinstance(self.max_decoded_audio_bytes, bool)
+            or not isinstance(self.max_decoded_audio_bytes, int)
+            or self.max_decoded_audio_bytes <= 0
+            or self.max_decoded_audio_bytes > sys.maxsize - 1
+        ):
+            raise ValueError(_POLICY_ERROR)
+        for source_bound in (
+            self.min_source_sample_rate,
+            self.max_source_sample_rate,
+            self.min_source_channels,
+            self.max_source_channels,
+        ):
+            if (
+                isinstance(source_bound, bool)
+                or not isinstance(source_bound, int)
+                or source_bound <= 0
+                or source_bound > sys.maxsize - 1
+            ):
+                raise ValueError(_POLICY_ERROR)
+        if (
+            self.min_source_sample_rate > self.max_source_sample_rate
+            or self.min_source_channels > self.max_source_channels
+        ):
+            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_source_metadata(
+        self,
+        frames: object,
+        sample_rate: object,
+        channels: object,
+    ) -> None:
+        """Validate source-container metadata before any decode transformation.
+
+        Args:
+            frames: Number of source frames reported by the container parser.
+            sample_rate: Source sample rate in Hz before resampling.
+            channels: Source channel count before downmixing.
+
+        Raises:
+            ValueError: If metadata is malformed or outside the source bounds.
+        """
+        if (
+            isinstance(frames, bool)
+            or not isinstance(frames, int)
+            or frames <= 0
+            or isinstance(sample_rate, bool)
+            or not isinstance(sample_rate, int)
+            or sample_rate < self.min_source_sample_rate
+            or sample_rate > self.max_source_sample_rate
+            or isinstance(channels, bool)
+            or not isinstance(channels, int)
+            or channels < self.min_source_channels
+            or channels > self.max_source_channels
+        ):
+            raise ValueError(_POLICY_ERROR)
+        try:
+            source_duration_seconds = float(frames) / float(sample_rate)
+        except (OverflowError, ValueError):
+            raise ValueError(_POLICY_ERROR) from None
+        if source_duration_seconds > float(self.max_duration_seconds):
+            raise ValueError(_POLICY_ERROR)
+
+    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, memory use,
+                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 audio.nbytes > self.max_decoded_audio_bytes
+            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_DECODED_AUDIO_BYTES",
+    "DEFAULT_MAX_DURATION_SECONDS",
+    "DEFAULT_MAX_ENCODED_FILE_BYTES",
+    "DEFAULT_MAX_SOURCE_CHANNELS",
+    "DEFAULT_MAX_SOURCE_SAMPLE_RATE",
+    "DEFAULT_MIN_SOURCE_CHANNELS",
+    "DEFAULT_MIN_SOURCE_SAMPLE_RATE",
+    "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..095050c1c 100644
--- a/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py
+++ b/services/analysis-engine/src/bandscope_analysis/separation/audio_separator.py
@@ -9,9 +9,16 @@
 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.
-- 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.
+- 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.
+- Empty, non-finite, or float32-overflowed model stems fail closed before they
+  can become successful silence or downstream rehearsal evidence.
+- Inference runs locally with no network access. Accelerator outputs cross back
+  to CPU before NumPy conversion so configured device execution cannot fail at
+  the device/host boundary. 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.
 - Does not log or persist raw audio, separated stems, or full source paths.
 - Fails with bounded, filename-scoped errors so callers can surface a safe
   failure without leaking local directory structure.
@@ -31,9 +38,13 @@
 import librosa
 import numpy as np
 
+from bandscope_analysis.audio_metadata import preflight_audio_metadata
+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,
 )
@@ -45,6 +56,7 @@
 # Demucs htdemucs emits these four sources; this is the canonical stem set.
 _STEM_ORDER: tuple[AudioStemName, ...] = ("vocals", "bass", "drums", "other")
 _EMPTY_RANGE_EPS = 1e-9
+_MODEL_OUTPUT_ERROR = "Stem separation produced invalid audio."
 
 
 def _contains_parent_path_segment(path: Path) -> bool:
@@ -63,7 +75,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 +87,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:
@@ -119,8 +136,8 @@ def _separate_signal(
         """Run the Demucs model on mono audio and return canonical mono stems.
 
         This is the single boundary to the neural model; it converts the mono
-        signal to the stereo tensor Demucs expects, applies the model on CPU, and
-        downmixes each source back to a mono float array.
+        signal to the stereo tensor Demucs expects, applies the model on the
+        configured device, and downmixes each source back to a mono host array.
         """
         model = self._load_model()
         sources = self._apply_model(model, audio)
@@ -172,7 +189,13 @@ def _apply_model(self, model: Any, audio: AudioStemArray) -> dict[str, np.ndarra
                 progress=False,
             )[0]
         out = out * ref_std + ref_mean
-        return {name: out[i].mean(0).numpy() for i, name in enumerate(model.sources)}
+        stems: dict[str, np.ndarray[Any, Any]] = {}
+        for index, name in enumerate(model.sources):
+            stem = out[index].mean(0)
+            if self.config.device != "cpu":
+                stem = stem.cpu()
+            stems[name] = stem.numpy()
+        return stems
 
     def _resolve_audio_file(self, audio_path: str | Path) -> Path:
         """Normalize and validate the selected source path."""
@@ -190,15 +213,17 @@ 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
+                preflight_audio_metadata(fileobj, self.resource_policy)
 
                 with warnings.catch_warnings():
                     warnings.filterwarnings(
@@ -214,16 +239,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."""
@@ -235,7 +263,12 @@ def _fit_length(self, audio: AudioStemArray, target_length: int) -> AudioStemArr
 
 
 def _as_float_array(values: object) -> AudioStemArray:
-    """Convert decoder and model output to a finite one-dimensional float array."""
-    array = np.ravel(np.asarray(values, dtype=np.float32))
-    finite = np.nan_to_num(array, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
-    return cast(AudioStemArray, finite)
+    """Convert one finite, non-empty decoder/model output into mono float32 audio."""
+    try:
+        with np.errstate(over="ignore", invalid="ignore"):
+            array = np.ravel(np.asarray(values, dtype=np.float32))
+    except (OverflowError, TypeError, ValueError) as error:
+        raise ValueError(_MODEL_OUTPUT_ERROR) from error
+    if array.size == 0 or not np.isfinite(array).all():
+        raise ValueError(_MODEL_OUTPUT_ERROR)
+    return cast(AudioStemArray, array)
diff --git a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
index 7fe5ae6f7..110222fa8 100644
--- a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
+++ b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py
@@ -12,18 +12,37 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from bandscope_analysis.audio_metadata import preflight_audio_metadata
+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.*"),
 )
+_SAFE_TEMPORAL_FAILURE_MESSAGES = frozenset(
+    {
+        "Audio file is too large for temporal analysis",
+        "Audio input violates the audio resource policy.",
+        "Expected numpy array from librosa.load",
+    }
+)
+_MISSING_AUDIO_MESSAGE = "Audio source is unavailable for temporal analysis."
+_GENERIC_TEMPORAL_FAILURE_MESSAGE = "Temporal analysis failed."
 # ponytail: assumes 4/4; upgrade to meter estimation or a madmom DBN if other meters matter.
 BEATS_PER_BAR = 4
 
@@ -56,11 +75,34 @@ def _estimate_downbeats(
     return [float(bt) for i, bt in enumerate(beat_times) if (i - best_phase) % beats_per_bar == 0]
 
 
+def _safe_temporal_failure_message(error: Exception) -> str:
+    """Return an allowlisted diagnostic without relaying decoder payload text."""
+    message = str(error)
+    if message in _SAFE_TEMPORAL_FAILURE_MESSAGES:
+        return message
+    return _GENERIC_TEMPORAL_FAILURE_MESSAGE
+
+
 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.
@@ -71,18 +113,18 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures:
         path = Path(audio_path)
         path_str = str(path)
         if not path.exists() or not path.is_file():
-            raise FileNotFoundError(f"Audio file not found: {path_str}")
+            raise FileNotFoundError(_MISSING_AUDIO_MESSAGE)
 
-        logger.info(f"Loading and decoding audio: {path_str}")
+        logger.info("Loading and decoding bounded local audio.")
 
         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
+                preflight_audio_metadata(fileobj, self.resource_policy)
 
                 with warnings.catch_warnings():
                     warnings.filterwarnings(
@@ -99,26 +141,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,
@@ -139,6 +180,6 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures:
                 "audio_path": path_str,
             }
 
-        except Exception as e:
-            logger.error(f"Failed to analyze audio {path_str}: {e}")
-            raise ValueError(f"Temporal analysis failed: {e}") from e
+        except Exception as error:
+            logger.error("Temporal analysis failed (%s).", type(error).__name__)
+            raise ValueError(_safe_temporal_failure_message(error)) from error
diff --git a/services/analysis-engine/src/bandscope_analysis/transcription/api.py b/services/analysis-engine/src/bandscope_analysis/transcription/api.py
index f2a732d31..b4090269f 100644
--- a/services/analysis-engine/src/bandscope_analysis/transcription/api.py
+++ b/services/analysis-engine/src/bandscope_analysis/transcription/api.py
@@ -10,6 +10,9 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from bandscope_analysis.audio_metadata import preflight_audio_metadata
+from bandscope_analysis.audio_resource_policy import AudioResourcePolicy
+
 TARGET_SR = 22050
 MAX_STEM_BYTES = 50 * 1024 * 1024
 MAX_TRANSCRIPTION_DURATION_SECONDS = 120
@@ -17,6 +20,12 @@
 HOP_LENGTH = 512
 MIN_NOTE_DURATION_SECONDS = 0.05
 MIN_SIGNAL_PEAK = 1e-5
+TRANSCRIPTION_RESOURCE_POLICY = AudioResourcePolicy(
+    max_encoded_file_bytes=MAX_STEM_BYTES,
+    target_sample_rate=TARGET_SR,
+    max_duration_seconds=MAX_TRANSCRIPTION_DURATION_SECONDS,
+    max_decoded_audio_bytes=(TARGET_SR * MAX_TRANSCRIPTION_DURATION_SECONDS + 1) * 8,
+)
 
 
 @dataclass
@@ -42,16 +51,20 @@ def transcribe_bass_stem(stem_data: bytes) -> list[NoteEvent]:
     if len(stem_data) > MAX_STEM_BYTES:
         raise ValueError("Stem data is too large for transcription.")
 
+    source = io.BytesIO(stem_data)
+    preflight_audio_metadata(source, TRANSCRIPTION_RESOURCE_POLICY)
     with warnings.catch_warnings():
         warnings.filterwarnings("ignore", category=DeprecationWarning, module=r"^audioread")
         y, sr = librosa.load(
-            io.BytesIO(stem_data),
+            source,
             sr=TARGET_SR,
             mono=True,
-            duration=MAX_TRANSCRIPTION_DURATION_SECONDS,
+            duration=TRANSCRIPTION_RESOURCE_POLICY.decode_probe_duration_seconds,
         )
 
-    y_array = np.asarray(y, dtype=np.float32)
+    y_array = np.asarray(
+        TRANSCRIPTION_RESOURCE_POLICY.validate_decoded_audio(y, sr), dtype=np.float32
+    )
     if y_array.size == 0 or float(np.max(np.abs(y_array))) < MIN_SIGNAL_PEAK:
         return []
 
diff --git a/services/analysis-engine/src/bandscope_analysis/youtube.py b/services/analysis-engine/src/bandscope_analysis/youtube.py
index c98f4e513..61c6bab0b 100644
--- a/services/analysis-engine/src/bandscope_analysis/youtube.py
+++ b/services/analysis-engine/src/bandscope_analysis/youtube.py
@@ -1,11 +1,35 @@
-"""
-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 duration must be a finite positive non-Boolean number when
+      present; malformed known-duration metadata fails closed before download.
+      Download-result duration is revalidated before success so changed
+      metadata cannot bypass the same 15-minute admission boundary.
+    - Announced ``filesize`` / ``filesize_approx`` values over the policy
+      ceiling reject the import before ``download=True``.
+    - The completed download path must resolve beneath this import's ``out_dir``
+      before post-download size checks, cleanup, or success metadata can use it.
+    - 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 +38,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 +51,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 +118,203 @@ 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 _download_error_result() -> Dict[str, Any]:
+    """Return the payload-safe generic import failure result."""
+    return {
+        "ok": False,
+        "error": {"code": "download_error", "message": YOUTUBE_IMPORT_FAILED_MESSAGE},
+    }
+
+
+def _reject_invalid_or_oversize_duration(info: dict[str, Any]) -> Dict[str, Any] | None:
+    """Validate announced duration before authorizing download work.
+
+    Args:
+        info: Metadata dictionary from yt-dlp extraction.
+
+    Returns:
+        A payload-safe failure for malformed/over-budget known duration, or
+        ``None`` when duration is absent or valid and within policy.
+    """
+    duration = info.get("duration")
+    if duration is None:
+        return None
+    if type(duration) not in (int, float):
+        return _download_error_result()
+    duration_seconds = float(duration)
+    if not math.isfinite(duration_seconds) or duration_seconds <= 0.0:
+        return _download_error_result()
+    if duration_seconds > DEFAULT_MAX_DURATION_SECONDS:
+        return {
+            "ok": False,
+            "error": {
+                "code": "duration_exceeded",
+                "message": "Video exceeds the 15-minute limit.",
+            },
+        }
+    return None
+
+
+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 +370,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:
@@ -137,21 +379,17 @@ def download_youtube_audio(url: str, out_dir: str) -> Dict[str, Any]:
             info = ydl.extract_info(url, download=False)
             if info is None:
                 raise Exception("Failed to extract info")
-            duration = info.get("duration")
-            if duration is not None and duration > 15 * 60:
-                return {
-                    "ok": False,
-                    "error": {
-                        "code": "duration_exceeded",
-                        "message": "Video exceeds the 15-minute limit.",
-                    },
-                }
+            duration_rejection = _reject_invalid_or_oversize_duration(info)
+            if duration_rejection is not None:
+                return duration_rejection
+            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:
                 raise Exception("Failed to extract info")
             actual_filepath = ydl.prepare_filename(info)
-
             actual_filepath = _find_downloaded_file(actual_filepath)
 
             if actual_filepath is None:
@@ -163,18 +401,24 @@ 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.",
-                    },
-                }
+            owned_filepath = _owned_file_path(actual_filepath, out_dir)
+            if owned_filepath is None:
+                return _download_error_result()
+            actual_filepath = owned_filepath
+
+            duration_rejection = _reject_invalid_or_oversize_duration(info)
+            if duration_rejection is not None:
+                _remove_owned_file(actual_filepath, out_dir)
+                return duration_rejection
+
+            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,13 +428,12 @@ 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:
-        return {
-            "ok": False,
-            "error": {"code": "download_error", "message": YOUTUBE_IMPORT_FAILED_MESSAGE},
-        }
+        return _download_error_result()
 
 
 def main() -> None:
diff --git a/services/analysis-engine/tests/test_audio_metadata.py b/services/analysis-engine/tests/test_audio_metadata.py
new file mode 100644
index 000000000..4c9b5c0ca
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_metadata.py
@@ -0,0 +1,93 @@
+"""Source-container metadata preflight regressions."""
+
+from __future__ import annotations
+
+import io
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from bandscope_analysis.audio_metadata import preflight_audio_metadata
+
+
+def _info(*, frames: int = 44_100, samplerate: int = 44_100, channels: int = 2) -> SimpleNamespace:
+    """Build the metadata subset consumed by the preflight boundary."""
+    return SimpleNamespace(frames=frames, samplerate=samplerate, channels=channels)
+
+
+@patch("bandscope_analysis.audio_metadata.soundfile.info")
+def test_preflight_accepts_metadata_and_rewinds_the_caller_handle(mock_info: object) -> None:
+    """A successful metadata probe leaves the decoder handle at its beginning."""
+    source = io.BytesIO(b"header-bytes")
+
+    def inspect(handle: io.BytesIO) -> SimpleNamespace:
+        """Consume a small header before returning parsed metadata."""
+        handle.read(3)
+        return _info()
+
+    mock_info.side_effect = inspect  # type: ignore[attr-defined]
+
+    preflight_audio_metadata(source)
+
+    assert source.tell() == 0
+
+
+@pytest.mark.parametrize(
+    ("info", "reason"),
+    [
+        (_info(frames=44_100 * 901), "audio resource policy"),
+        (_info(samplerate=7_999), "audio resource policy"),
+        (_info(channels=3), "audio resource policy"),
+    ],
+)
+@patch("bandscope_analysis.audio_metadata.soundfile.info")
+def test_preflight_rejects_untrusted_source_metadata(
+    mock_info: object,
+    info: SimpleNamespace,
+    reason: str,
+) -> None:
+    """Source duration, rate, and channel bounds fail before PCM decode."""
+    mock_info.return_value = info  # type: ignore[attr-defined]
+
+    with pytest.raises(ValueError, match=reason):
+        preflight_audio_metadata(io.BytesIO(b"header"))
+
+
+@patch(
+    "bandscope_analysis.audio_metadata.soundfile.info",
+    side_effect=RuntimeError("decoder detail"),
+)
+def test_preflight_maps_parser_failures_to_payload_free_policy_error(_mock_info: object) -> None:
+    """Container parser failures do not leak decoder details."""
+    with pytest.raises(ValueError, match="audio resource policy") as error:
+        preflight_audio_metadata(io.BytesIO(b"bad-header"))
+
+    assert "decoder detail" not in str(error.value)
+
+
+@patch("bandscope_analysis.audio_metadata.soundfile.info")
+def test_preflight_maps_rewind_failures_to_payload_free_policy_error(mock_info: object) -> None:
+    """A handle that cannot rewind after probing cannot reach a decoder."""
+
+    class SeekFailsAfterProbe(io.BytesIO):
+        """Fail only when the metadata boundary tries to rewind the handle."""
+
+        def __init__(self) -> None:
+            """Initialize the caller-owned byte handle and seek counter."""
+            super().__init__(b"header")
+            self.seek_count = 0
+
+        def seek(self, *args: object, **kwargs: object) -> int:
+            """Reject the second seek, which is the post-probe rewind."""
+            self.seek_count += 1
+            if self.seek_count == 2:
+                raise OSError("rewind failed")
+            return super().seek(*args, **kwargs)
+
+    mock_info.return_value = _info()  # type: ignore[attr-defined]
+
+    with pytest.raises(ValueError, match="audio resource policy") as error:
+        preflight_audio_metadata(SeekFailsAfterProbe())
+
+    assert "rewind failed" not in str(error.value)
diff --git a/services/analysis-engine/tests/test_audio_model_output_policy.py b/services/analysis-engine/tests/test_audio_model_output_policy.py
new file mode 100644
index 000000000..863b5418c
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_model_output_policy.py
@@ -0,0 +1,39 @@
+"""Regression tests for fail-closed source-separation model output."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from bandscope_analysis.separation.audio_separator import _as_float_array
+
+
+@pytest.mark.parametrize(
+    "values",
+    [
+        np.array([], dtype=np.float32),
+        np.array([np.nan], dtype=np.float32),
+        np.array([np.inf], dtype=np.float32),
+        np.array([np.finfo(np.float64).max], dtype=np.float64),
+    ],
+)
+def test_model_output_rejects_empty_nonfinite_or_float32_overflow(values: np.ndarray) -> None:
+    """Malformed model stems must fail instead of becoming successful silence."""
+    with pytest.raises(ValueError, match=r"^Stem separation produced invalid audio\.$"):
+        _as_float_array(values)
+
+
+def test_model_output_wraps_non_numeric_conversion_errors() -> None:
+    """Non-numeric model output must fail with the stable payload-free error."""
+    with pytest.raises(ValueError, match=r"^Stem separation produced invalid audio\.$"):
+        _as_float_array(object())
+
+
+def test_model_output_preserves_valid_finite_samples() -> None:
+    """Valid model samples remain finite float32 audio with their original values."""
+    values = np.array([0.25, -0.5, 0.75], dtype=np.float64)
+
+    result = _as_float_array(values)
+
+    assert result.dtype == np.float32
+    assert np.array_equal(result, values.astype(np.float32))
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..2d52f44a5
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy.py
@@ -0,0 +1,178 @@
+"""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,
+    DEFAULT_MAX_SOURCE_CHANNELS,
+    DEFAULT_MAX_SOURCE_SAMPLE_RATE,
+    DEFAULT_MIN_SOURCE_CHANNELS,
+    DEFAULT_MIN_SOURCE_SAMPLE_RATE,
+    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
+    assert DEFAULT_AUDIO_RESOURCE_POLICY.max_decoded_audio_bytes == 44_100 * 15 * 60 * 8
+
+
+@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
+
+
+def test_source_metadata_accepts_the_published_bounds() -> None:
+    """Source metadata accepts the inclusive rate, channel, and duration bounds."""
+    policy = AudioResourcePolicy(max_duration_seconds=15 * 60)
+
+    policy.validate_source_metadata(
+        frames=DEFAULT_MAX_SOURCE_SAMPLE_RATE * 15 * 60,
+        sample_rate=DEFAULT_MAX_SOURCE_SAMPLE_RATE,
+        channels=DEFAULT_MAX_SOURCE_CHANNELS,
+    )
+    policy.validate_source_metadata(
+        frames=DEFAULT_MIN_SOURCE_SAMPLE_RATE,
+        sample_rate=DEFAULT_MIN_SOURCE_SAMPLE_RATE,
+        channels=DEFAULT_MIN_SOURCE_CHANNELS,
+    )
+
+
+@pytest.mark.parametrize(
+    ("frames", "sample_rate", "channels"),
+    [
+        (DEFAULT_MAX_SOURCE_SAMPLE_RATE * (15 * 60 + 1), 44_100, 2),
+        (44_100, DEFAULT_MIN_SOURCE_SAMPLE_RATE - 1, 2),
+        (44_100, DEFAULT_MAX_SOURCE_SAMPLE_RATE + 1, 2),
+        (44_100, 44_100, DEFAULT_MAX_SOURCE_CHANNELS + 1),
+        (44_100, 44_100, DEFAULT_MIN_SOURCE_CHANNELS - 1),
+        (0, 44_100, 2),
+        (44_100, True, 2),
+        (44_100, 44_100, True),
+        (10**400, 44_100, 2),
+    ],
+)
+def test_source_metadata_fails_closed_before_decode(
+    frames: object,
+    sample_rate: object,
+    channels: object,
+) -> None:
+    """Overlong and malformed source metadata cannot reach a decoder."""
+    with pytest.raises(ValueError, match="audio resource policy"):
+        DEFAULT_AUDIO_RESOURCE_POLICY.validate_source_metadata(frames, sample_rate, channels)
+
+
+@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_rejects_buffer_above_memory_budget() -> None:
+    """A decoder cannot hide excessive memory behind an allowed sample count."""
+    policy = AudioResourcePolicy(
+        target_sample_rate=8,
+        max_duration_seconds=1.0,
+        max_decoded_audio_bytes=16,
+    )
+    audio = np.zeros(4, dtype=np.float64)
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        policy.validate_decoded_audio(audio, 8)
+
+
+def test_decoded_audio_accepts_exact_memory_boundary() -> None:
+    """A finite canonical buffer exactly at the memory ceiling is accepted."""
+    policy = AudioResourcePolicy(
+        target_sample_rate=8,
+        max_duration_seconds=1.0,
+        max_decoded_audio_bytes=32,
+    )
+    audio = np.zeros(8, dtype=np.float32)
+
+    assert policy.validate_decoded_audio(audio, 8) is audio
+
+
+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")},
+        {"max_decoded_audio_bytes": 0},
+        {"max_decoded_audio_bytes": True},
+        {"min_source_sample_rate": 0},
+        {"max_source_channels": True},
+        {"min_source_sample_rate": 48_000, "max_source_sample_rate": 44_100},
+        {"min_source_channels": 2, "max_source_channels": 1},
+    ],
+)
+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},
+        {"max_decoded_audio_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..f45e5d454
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy_coverage_regressions.py
@@ -0,0 +1,74 @@
+"""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..1f58720f1
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_resource_policy_integration.py
@@ -0,0 +1,241 @@
+"""Cross-boundary regressions for canonical local-audio resource admission."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+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")
+    monkeypatch.setattr(
+        "bandscope_analysis.temporal.analyzer.preflight_audio_metadata",
+        lambda *_args, **_kwargs: None,
+    )
+    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
+
+
+@pytest.mark.parametrize(
+    "metadata",
+    [
+        SimpleNamespace(frames=44_100 * 901, samplerate=44_100, channels=2),
+        SimpleNamespace(frames=44_100, samplerate=7_999, channels=2),
+        SimpleNamespace(frames=44_100, samplerate=44_100, channels=3),
+    ],
+)
+def test_temporal_rejects_source_metadata_before_librosa_decode(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+    metadata: SimpleNamespace,
+) -> None:
+    """Temporal analysis must inspect source metadata before resampling or truncation."""
+    import librosa
+
+    source = tmp_path / "source-metadata.wav"
+    source.write_bytes(b"bounded")
+    monkeypatch.setattr(
+        "bandscope_analysis.audio_metadata.soundfile.info",
+        lambda _fileobj: metadata,
+    )
+    load_mock = Mock(side_effect=AssertionError("source metadata must be checked first"))
+    monkeypatch.setattr(librosa, "load", load_mock)
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        TemporalAnalyzer().analyze(source)
+
+    load_mock.assert_not_called()
+
+
+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")
+    monkeypatch.setattr(
+        "bandscope_analysis.separation.audio_separator.preflight_audio_metadata",
+        lambda *_args, **_kwargs: None,
+    )
+    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
+
+
+@pytest.mark.parametrize(
+    "metadata",
+    [
+        SimpleNamespace(frames=44_100 * 901, samplerate=44_100, channels=2),
+        SimpleNamespace(frames=44_100, samplerate=7_999, channels=2),
+        SimpleNamespace(frames=44_100, samplerate=44_100, channels=3),
+    ],
+)
+def test_stem_decoder_rejects_source_metadata_before_librosa_decode(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+    metadata: SimpleNamespace,
+) -> None:
+    """Stem separation must inspect source metadata before mono conversion or model work."""
+    import librosa
+
+    source = tmp_path / "source-metadata.wav"
+    source.write_bytes(b"bounded")
+    monkeypatch.setattr(
+        "bandscope_analysis.audio_metadata.soundfile.info",
+        lambda _fileobj: metadata,
+    )
+    load_mock = Mock(side_effect=AssertionError("source metadata must be checked first"))
+    monkeypatch.setattr(librosa, "load", load_mock)
+
+    separator = AudioStemSeparator(AudioSeparationConfig(max_file_bytes=100))
+    with pytest.raises(ValueError, match="audio resource policy"):
+        separator.separate(source)
+
+    load_mock.assert_not_called()
+
+
+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(
+        "bandscope_analysis.separation.audio_separator.preflight_audio_metadata",
+        lambda *_args, **_kwargs: None,
+    )
+    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_audio_separator_device_boundary.py b/services/analysis-engine/tests/test_audio_separator_device_boundary.py
new file mode 100644
index 000000000..e8545876b
--- /dev/null
+++ b/services/analysis-engine/tests/test_audio_separator_device_boundary.py
@@ -0,0 +1,109 @@
+"""Device-boundary regressions for local Demucs separation."""
+
+from __future__ import annotations
+
+import sys
+from types import ModuleType
+
+import numpy as np
+import pytest
+
+from bandscope_analysis.separation.audio_separator import AudioSeparationConfig, AudioStemSeparator
+
+
+class _FakeModel:
+    """Expose the canonical Demucs source order used by production."""
+
+    sources = ["drums", "bass", "other", "vocals"]
+
+
+class _DeviceTensor:
+    """Minimal tensor that refuses NumPy conversion until moved to CPU."""
+
+    def __init__(self, array: np.ndarray, *, on_cpu: bool) -> None:
+        self.array = np.asarray(array, dtype=np.float32)
+        self.on_cpu = on_cpu
+
+    def float(self) -> "_DeviceTensor":
+        return _DeviceTensor(self.array.astype(np.float32), on_cpu=self.on_cpu)
+
+    def mean(self, axis: int | None = None) -> float | "_DeviceTensor":
+        value = self.array.mean(axis=axis)
+        if axis is None:
+            return float(value)
+        return _DeviceTensor(np.asarray(value, dtype=np.float32), on_cpu=self.on_cpu)
+
+    def std(self) -> float:
+        return float(self.array.std())
+
+    def cpu(self) -> "_DeviceTensor":
+        return _DeviceTensor(self.array, on_cpu=True)
+
+    def numpy(self) -> np.ndarray:
+        if not self.on_cpu:
+            raise RuntimeError("can't convert cuda tensor to numpy")
+        return self.array
+
+    def __getitem__(self, key: object) -> "_DeviceTensor":
+        return _DeviceTensor(self.array[key], on_cpu=self.on_cpu)
+
+    def __add__(self, value: float) -> "_DeviceTensor":
+        return _DeviceTensor(self.array + value, on_cpu=self.on_cpu)
+
+    def __sub__(self, value: float) -> "_DeviceTensor":
+        return _DeviceTensor(self.array - value, on_cpu=self.on_cpu)
+
+    def __mul__(self, value: float) -> "_DeviceTensor":
+        return _DeviceTensor(self.array * value, on_cpu=self.on_cpu)
+
+    def __truediv__(self, value: float) -> "_DeviceTensor":
+        return _DeviceTensor(self.array / value, on_cpu=self.on_cpu)
+
+
+class _NoGrad:
+    def __enter__(self) -> None:
+        return None
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+
+def test_apply_model_moves_device_output_to_cpu_before_numpy(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """GPU-selected separation must cross the device boundary before NumPy conversion."""
+    calls: dict[str, object] = {}
+    fake_torch = ModuleType("torch")
+    fake_torch.from_numpy = lambda array: _DeviceTensor(array, on_cpu=True)  # type: ignore[attr-defined]
+    fake_torch.no_grad = _NoGrad  # type: ignore[attr-defined]
+
+    def fake_apply_model(
+        model: _FakeModel,
+        batch: _DeviceTensor,
+        *,
+        device: str,
+        split: bool,
+        overlap: float,
+        progress: bool,
+    ) -> _DeviceTensor:
+        calls.update(device=device, split=split, overlap=overlap, progress=progress)
+        source_values = np.arange(len(model.sources), dtype=np.float32).reshape(-1, 1, 1)
+        separated = np.broadcast_to(source_values, (len(model.sources), 2, 4)).copy()
+        return _DeviceTensor(separated[None], on_cpu=False)
+
+    demucs_module = ModuleType("demucs")
+    apply_module = ModuleType("demucs.apply")
+    apply_module.apply_model = fake_apply_model  # type: ignore[attr-defined]
+    demucs_module.apply = apply_module  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "torch", fake_torch)
+    monkeypatch.setitem(sys.modules, "demucs", demucs_module)
+    monkeypatch.setitem(sys.modules, "demucs.apply", apply_module)
+
+    audio = np.array([0.0, 1.0, -1.0, 0.5], dtype=np.float32)
+    separator = AudioStemSeparator(AudioSeparationConfig(device="cuda", overlap=0.375))
+
+    result = separator._apply_model(_FakeModel(), audio)
+
+    assert calls == {"device": "cuda", "split": True, "overlap": 0.375, "progress": False}
+    assert set(result) == set(_FakeModel.sources)
+    assert all(stem.shape == (4,) for stem in result.values())
diff --git a/services/analysis-engine/tests/test_separation.py b/services/analysis-engine/tests/test_separation.py
index f8e098521..649fb0f23 100644
--- a/services/analysis-engine/tests/test_separation.py
+++ b/services/analysis-engine/tests/test_separation.py
@@ -465,6 +465,10 @@ def test_audio_stem_separator_rejects_empty_decoder_output(
     """Ensure empty decoder output fails safely."""
     audio_path = tmp_path / "empty.wav"
     audio_path.write_bytes(b"placeholder")
+    monkeypatch.setattr(
+        "bandscope_analysis.separation.audio_separator.preflight_audio_metadata",
+        lambda *_args, **_kwargs: None,
+    )
     monkeypatch.setattr(
         "bandscope_analysis.separation.audio_separator.librosa.load",
         lambda *args, **kwargs: (np.array([], dtype=np.float32), 8_000),
@@ -481,6 +485,10 @@ def test_audio_stem_separator_redacts_decoder_exceptions(
     """Ensure decoder failures are surfaced without full local paths."""
     audio_path = tmp_path / "broken.wav"
     audio_path.write_bytes(b"placeholder")
+    monkeypatch.setattr(
+        "bandscope_analysis.separation.audio_separator.preflight_audio_metadata",
+        lambda *_args, **_kwargs: None,
+    )
 
     def fail_decode(*args, **kwargs):
         raise RuntimeError(f"decoder failed under {tmp_path}")
diff --git a/services/analysis-engine/tests/test_stem_separation_logging_privacy.py b/services/analysis-engine/tests/test_stem_separation_logging_privacy.py
new file mode 100644
index 000000000..8d7d2d7b1
--- /dev/null
+++ b/services/analysis-engine/tests/test_stem_separation_logging_privacy.py
@@ -0,0 +1,124 @@
+"""Regression tests for stem-separation logging privacy."""
+
+import logging
+
+import pytest
+
+import bandscope_analysis.api as analysis_api
+
+
+class _ResultQueue:
+    """Capture the worker result without starting a multiprocessing queue."""
+
+    def __init__(self) -> None:
+        self.items: list[tuple[object, object]] = []
+
+    def put(self, item: tuple[object, object]) -> None:
+        """Record one result emitted by the worker."""
+        self.items.append(item)
+
+
+class _FailingSeparator:
+    """Raise dependency-controlled sensitive text from the separator boundary."""
+
+    def separate(self, source_path: str) -> dict[str, object]:
+        """Simulate a dependency failure after receiving an authorized source path."""
+        raise RuntimeError(
+            f"decoder failed for {source_path} /Users/Alice/private-song.wav token=super-secret"
+        )
+
+
+def _local_audio_request() -> dict[str, object]:
+    """Return a valid local-audio request without cache or temporary-path authority."""
+    return {
+        "sourceKind": "local_audio",
+        "projectId": "privacy-regression",
+        "sourceLabel": "private-song.wav",
+        "roleFocus": ["bass-guitar"],
+        "localSource": {
+            "sourcePath": "/private/customer/Alice/session.wav",
+            "fileName": "private-song.wav",
+            "extension": "wav",
+            "fileSizeBytes": 1024,
+        },
+    }
+
+
+def _assert_payload_free_log(caplog: pytest.LogCaptureFixture) -> None:
+    """Require routine logs to omit dependency payloads and exception tracebacks."""
+    assert "/private/customer/Alice/session.wav" not in caplog.text
+    assert "/Users/Alice/private-song.wav" not in caplog.text
+    assert "private-song.wav token=super-secret" not in caplog.text
+    assert "super-secret" not in caplog.text
+    assert all(record.exc_info is None for record in caplog.records)
+
+
+def test_stem_worker_failure_log_omits_dependency_payload_and_traceback(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Routine worker diagnostics must not retain dependency payloads or tracebacks."""
+    result_queue = _ResultQueue()
+    source_path = "/private/customer/Alice/session.wav"
+
+    monkeypatch.setattr(analysis_api, "AudioStemSeparator", _FailingSeparator)
+    caplog.set_level(logging.ERROR, logger=analysis_api.__name__)
+
+    analysis_api._stem_separation_worker(source_path, result_queue)
+
+    assert result_queue.items == [
+        ("runtime_error", "Runtime error occurred during stem separation.")
+    ]
+    assert "Stem separation failed with a runtime error." in caplog.text
+    _assert_payload_free_log(caplog)
+
+
+def test_analysis_job_stem_failure_log_omits_dependency_payload_and_traceback(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Parent orchestration failure logs must keep dependency details out of routine logs."""
+    sensitive_detail = (
+        "decode failed for /private/customer/Alice/session.wav "
+        "/Users/Alice/private-song.wav token=super-secret"
+    )
+
+    def fail_features(_request: analysis_api.AnalysisJobRequest) -> None:
+        raise ValueError(sensitive_detail)
+
+    monkeypatch.setattr(analysis_api, "_build_local_audio_features", fail_features)
+    caplog.set_level(logging.ERROR, logger=analysis_api.__name__)
+
+    updates = analysis_api.run_analysis_job_updates(
+        "job-privacy",
+        _local_audio_request(),
+        "2026-08-20T00:00:00Z",
+    )
+
+    assert updates[-1]["state"] == "failed"
+    assert updates[-1]["error"] == {
+        "code": "engine_unavailable",
+        "message": "Stem separation failed",
+    }
+    assert "Stem separation failed before analysis job completion." in caplog.text
+    _assert_payload_free_log(caplog)
+
+
+def test_api_logger_preserves_unrelated_exception_traceback(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Privacy redaction must not erase traceback evidence from unrelated API diagnostics."""
+    caplog.set_level(logging.ERROR, logger=analysis_api.__name__)
+
+    try:
+        raise RuntimeError("non-sensitive diagnostic sentinel")
+    except RuntimeError:
+        analysis_api.logger.exception("Unrelated analysis API diagnostic.")
+
+    records = [
+        record
+        for record in caplog.records
+        if record.getMessage() == "Unrelated analysis API diagnostic."
+    ]
+    assert len(records) == 1
+    assert records[0].exc_info is not None
diff --git a/services/analysis-engine/tests/test_temporal.py b/services/analysis-engine/tests/test_temporal.py
index 6ce90ae1c..16c7f7034 100644
--- a/services/analysis-engine/tests/test_temporal.py
+++ b/services/analysis-engine/tests/test_temporal.py
@@ -46,9 +46,9 @@ def test_temporal_analyzer_basic(dummy_audio_file: Path) -> None:
 
 
 def test_temporal_analyzer_file_not_found() -> None:
-    """Test that analyzer raises appropriate error for missing files."""
+    """Test that analyzer raises a payload-safe error for missing files."""
     analyzer = TemporalAnalyzer()
-    with pytest.raises(FileNotFoundError, match="Audio file not found"):
+    with pytest.raises(FileNotFoundError, match="Audio source is unavailable"):
         analyzer.analyze("nonexistent_file.wav")
 
 
@@ -62,7 +62,7 @@ def test_temporal_analyzer_missing_file_does_not_call_decoder(
     monkeypatch.setattr(librosa, "load", load_mock)
 
     analyzer = TemporalAnalyzer()
-    with pytest.raises(FileNotFoundError, match="Audio file not found"):
+    with pytest.raises(FileNotFoundError, match="Audio source is unavailable"):
         analyzer.analyze("nonexistent_file.wav")
     load_mock.assert_not_called()
 
@@ -77,7 +77,7 @@ def test_temporal_analyzer_directory_does_not_call_decoder(
     load_mock = Mock(side_effect=AssertionError("librosa.load should not be called"))
     monkeypatch.setattr(librosa, "load", load_mock)
 
-    with pytest.raises(FileNotFoundError, match="Audio file not found"):
+    with pytest.raises(FileNotFoundError, match="Audio source is unavailable"):
         TemporalAnalyzer().analyze(tmp_path)
     load_mock.assert_not_called()
 
@@ -94,7 +94,7 @@ def fake_load(*args, **kwargs):
     monkeypatch.setattr(librosa, "load", fake_load)
 
     test_wav = tmp_path / "test.wav"
-    test_wav.write_bytes(b"dummy")
+    sf.write(test_wav, np.zeros(4_000, dtype=np.float32), 44_100)
 
     with pytest.raises(ValueError, match="Expected numpy array"):
         TemporalAnalyzer().analyze(test_wav)
@@ -104,7 +104,7 @@ def test_temporal_analyzer_exception_handling(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
 ) -> None:
-    """Ensure temporal analyzer catches general exceptions and raises ValueError."""
+    """Ensure arbitrary decoder exception payloads are not relayed to callers."""
     import librosa
 
     from bandscope_analysis.temporal.analyzer import TemporalAnalyzer
@@ -115,10 +115,11 @@ def fake_load(*args: object, **kwargs: object) -> tuple[np.ndarray, int]:
     monkeypatch.setattr(librosa, "load", fake_load)
 
     test_wav = tmp_path / "test.wav"
-    test_wav.write_bytes(b"dummy")
+    sf.write(test_wav, np.zeros(4_000, dtype=np.float32), 44_100)
 
-    with pytest.raises(ValueError, match="Temporal analysis failed: Mocked general error"):
+    with pytest.raises(ValueError, match=r"^Temporal analysis failed\.$") as exc_info:
         TemporalAnalyzer().analyze(test_wav)
+    assert "Mocked general error" not in str(exc_info.value)
 
 
 def test_temporal_analyzer_rejects_oversized_file(monkeypatch, tmp_path: Path) -> None:
@@ -128,7 +129,7 @@ def test_temporal_analyzer_rejects_oversized_file(monkeypatch, tmp_path: Path) -
     from bandscope_analysis.temporal import analyzer as analyzer_module
 
     test_wav = tmp_path / "large.wav"
-    test_wav.write_bytes(b"1234")
+    sf.write(test_wav, np.zeros(4_000, dtype=np.float32), 44_100)
 
     monkeypatch.setattr(analyzer_module, "MAX_AUDIO_FILE_BYTES", 1)
 
@@ -147,7 +148,7 @@ def test_temporal_analyzer_uses_duration_limit(monkeypatch, tmp_path: Path) -> N
     import librosa
 
     test_wav = tmp_path / "bounded.wav"
-    test_wav.write_bytes(b"1234")
+    sf.write(test_wav, np.zeros(4_000, dtype=np.float32), 44_100)
     captured_kwargs: dict[str, object] = {}
 
     def fake_load(path, **kwargs):
@@ -178,7 +179,7 @@ def test_temporal_analyzer_does_not_suppress_unrelated_loader_warnings(
     import librosa
 
     test_wav = tmp_path / "test.wav"
-    test_wav.write_bytes(b"dummy")
+    sf.write(test_wav, np.zeros(4_000, dtype=np.float32), 44_100)
 
     def fake_load(*args: object, **kwargs: object) -> tuple[np.ndarray, int]:
         warnings.warn("unrelated downstream warning", FutureWarning, stacklevel=2)
diff --git a/services/analysis-engine/tests/test_temporal_error_privacy.py b/services/analysis-engine/tests/test_temporal_error_privacy.py
new file mode 100644
index 000000000..ea0c6519f
--- /dev/null
+++ b/services/analysis-engine/tests/test_temporal_error_privacy.py
@@ -0,0 +1,57 @@
+"""Privacy regressions for temporal-analysis failure diagnostics."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import numpy as np
+import pytest
+import soundfile as sf
+
+from bandscope_analysis.temporal import TemporalAnalyzer
+
+
+def test_missing_temporal_source_does_not_disclose_local_path(tmp_path: Path) -> None:
+    """Missing-file failures must not echo an absolute customer path to callers."""
+    sensitive_path = tmp_path / "private-customer-session" / "unreleased-song.wav"
+
+    with pytest.raises(FileNotFoundError) as exc_info:
+        TemporalAnalyzer().analyze(sensitive_path)
+
+    message = str(exc_info.value)
+    assert message == "Audio source is unavailable for temporal analysis."
+    assert str(sensitive_path) not in message
+    assert "unreleased-song.wav" not in message
+
+
+def test_decoder_failure_redacts_source_path_and_decoder_payload(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Decoder diagnostics must remain useful without logging customer path/payload data."""
+    import librosa
+
+    sensitive_path = tmp_path / "private-customer-session" / "unreleased-song.wav"
+    sensitive_path.parent.mkdir()
+    sf.write(sensitive_path, np.zeros(4_000, dtype=np.float32), 44_100)
+    decoder_payload = "decoder exposed /private/customer/token-shaped-audio-name.wav"
+
+    def fail_decode(*args: object, **kwargs: object) -> tuple[object, int]:
+        raise RuntimeError(decoder_payload)
+
+    monkeypatch.setattr(librosa, "load", fail_decode)
+    caplog.set_level(logging.INFO, logger="bandscope_analysis.temporal.analyzer")
+
+    with pytest.raises(ValueError) as exc_info:
+        TemporalAnalyzer().analyze(sensitive_path)
+
+    message = str(exc_info.value)
+    assert message == "Temporal analysis failed."
+    assert str(sensitive_path) not in message
+    assert decoder_payload not in message
+    assert str(sensitive_path) not in caplog.text
+    assert "unreleased-song.wav" not in caplog.text
+    assert decoder_payload not in caplog.text
+    assert "RuntimeError" in caplog.text
diff --git a/services/analysis-engine/tests/test_transcription.py b/services/analysis-engine/tests/test_transcription.py
index f9b55af93..80eb126ff 100644
--- a/services/analysis-engine/tests/test_transcription.py
+++ b/services/analysis-engine/tests/test_transcription.py
@@ -4,8 +4,11 @@
 
 import io
 from dataclasses import dataclass
+from types import SimpleNamespace
+from unittest.mock import Mock
 
 import numpy as np
+import pytest
 import soundfile as sf
 
 from bandscope_analysis.transcription import api as transcription_api
@@ -62,6 +65,32 @@ def test_transcribe_bass_stem_rejects_oversized_input(monkeypatch) -> None:
         transcribe_bass_stem(b"abc")
 
 
+@pytest.mark.parametrize(
+    "metadata",
+    [
+        SimpleNamespace(frames=22050 * 121, samplerate=22050, channels=2),
+        SimpleNamespace(frames=22050, samplerate=7_999, channels=2),
+        SimpleNamespace(frames=22050, samplerate=22050, channels=3),
+    ],
+)
+def test_transcribe_bass_stem_rejects_source_metadata_before_decode(
+    monkeypatch: pytest.MonkeyPatch,
+    metadata: SimpleNamespace,
+) -> None:
+    """Bass transcription must validate source duration, rate, and channels before librosa."""
+    monkeypatch.setattr(
+        "bandscope_analysis.audio_metadata.soundfile.info",
+        lambda _fileobj: metadata,
+    )
+    load_mock = Mock(side_effect=AssertionError("source metadata must be checked first"))
+    monkeypatch.setattr(transcription_api.librosa, "load", load_mock)
+
+    with pytest.raises(ValueError, match="audio resource policy"):
+        transcribe_bass_stem(b"not-a-real-wav")
+
+    load_mock.assert_not_called()
+
+
 def test_transcribe_bass_stem_wraps_pitch_tracking_parameter_errors(monkeypatch) -> None:
     """Return a stable ValueError when pYIN rejects decoded audio parameters."""
     stem_data = _render_bass_sequence([ExpectedNote("E2", 0.0, 0.45)])
diff --git a/services/analysis-engine/tests/test_youtube.py b/services/analysis-engine/tests/test_youtube.py
index 5531ac9d5..0ae449aa9 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,20 +99,23 @@ def test_download_youtube_audio_success(
         "id": "abc123DEF45",
         "title": "Test Video",
         "duration": 60,
+        "filesize": True,
+        "filesize_approx": float("nan"),
     }
+    out_dir = str(Path("/tmp").resolve())
     mock_ydl.extract_info.return_value = mock_info
-    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.webm"
+    mock_ydl.prepare_filename.return_value = f"{out_dir}/abc123DEF45.webm"
     mock_exists.return_value = True
     mock_getsize.return_value = 10 * 1024 * 1024
 
     input_url = "https://youtube.com/watch?v=abc123DEF45"
-    result = download_youtube_audio(input_url, "/tmp")
+    result = download_youtube_audio(input_url, out_dir)
 
     assert result["ok"] is True
     assert result["metadata"]["id"] == "abc123DEF45"
     assert result["metadata"]["title"] == "Test Video"
     assert result["metadata"]["duration"] == 60
-    assert result["metadata"]["filepath"] == "/tmp/abc123DEF45.webm"
+    assert result["metadata"]["filepath"] == f"{out_dir}/abc123DEF45.webm"
 
     # Assert that YoutubeDL was initialized with the correct options
     mock_ydl_class.assert_called_once()
@@ -114,6 +127,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
@@ -145,21 +160,22 @@ def test_download_youtube_audio_converted_extension(
         "title": "Test Video",
         "duration": 60,
     }
+    out_dir = str(Path("/tmp").resolve())
     mock_ydl.extract_info.return_value = mock_info
-    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.webm"
+    mock_ydl.prepare_filename.return_value = f"{out_dir}/abc123DEF45.webm"
 
     # os.path.exists returns False for .webm, but True for the converted .opus.
     def exists_side_effect(path: str) -> bool:
         """Mock exists function to simulate converted extension file presence."""
-        return path == "/tmp/abc123DEF45.opus"
+        return path == f"{out_dir}/abc123DEF45.opus"
 
     mock_exists.side_effect = exists_side_effect
     mock_getsize.return_value = 10 * 1024 * 1024
 
-    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", out_dir)
 
     assert result["ok"] is True
-    assert result["metadata"]["filepath"] == "/tmp/abc123DEF45.opus"
+    assert result["metadata"]["filepath"] == f"{out_dir}/abc123DEF45.opus"
 
 
 @patch("bandscope_analysis.youtube.os.path.exists")
@@ -273,6 +289,50 @@ 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
+    out_dir = str(Path("/tmp").resolve())
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
+    mock_ydl.prepare_filename.return_value = f"{out_dir}/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_getsize.return_value = 60 * 1024 * 1024
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", out_dir)
+
+    assert result["ok"] is True
+    assert result["metadata"]["filepath"] == f"{out_dir}/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,18 +343,315 @@ 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
+    out_dir = str(Path("/tmp").resolve())
+    mock_ydl.extract_info.return_value = {"id": "abc123DEF45", "duration": 10 * 60}
+    mock_ydl.prepare_filename.return_value = f"{out_dir}/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_getsize.return_value = DEFAULT_MAX_ENCODED_FILE_BYTES + 1
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", out_dir)
+    assert result["ok"] is False
+    assert result["error"]["code"] == "size_exceeded"
+    assert result["error"]["message"] == YOUTUBE_SIZE_EXCEEDED_MESSAGE
+    mock_remove.assert_called_with(f"{out_dir}/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 = 51 * 1024 * 1024
+    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"
-    mock_remove.assert_called_with("/tmp/abc123DEF45.m4a")
+    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:
@@ -326,38 +683,35 @@ def test_main_block(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixtu
 
 
 def test_module_execution(
-    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    tmp_path: Path,
 ) -> None:
-    """Test the if __name__ == '__main__' block using runpy."""
+    """Test module execution against a real owned output path without network I/O."""
     import runpy
 
     import bandscope_analysis.youtube
 
+    downloaded_path = tmp_path / "abc123DEF45.m4a"
+    downloaded_path.write_bytes(b"test-audio")
     test_args = [
         "youtube.py",
         "--url",
         "https://youtube.com/watch?v=abc123DEF45",
         "--out-dir",
-        "/tmp",
+        str(tmp_path),
     ]
     monkeypatch.setattr(sys, "argv", test_args)
 
-    # Mock yt_dlp so runpy doesn't actually download
+    # Mock only the downloader/network boundary. Real filesystem semantics are
+    # required so the completed-path ownership check remains exercised.
     mock_yt_dlp = MagicMock()
     mock_ydl = MagicMock()
     mock_yt_dlp.YoutubeDL.return_value.__enter__.return_value = mock_ydl
     mock_ydl.extract_info.return_value = {"id": "abc123DEF45"}
-    mock_ydl.prepare_filename.return_value = "/tmp/abc123DEF45.m4a"
+    mock_ydl.prepare_filename.return_value = str(downloaded_path)
     monkeypatch.setitem(sys.modules, "yt_dlp", mock_yt_dlp)
 
-    # Mock os to ensure runpy uses our mocked filesystem methods
-    mock_os = MagicMock()
-    # Keep some essential attributes
-    mock_os.path = MagicMock()
-    mock_os.path.exists.return_value = True
-    mock_os.path.getsize.return_value = 10 * 1024 * 1024
-    monkeypatch.setitem(sys.modules, "os", mock_os)
-
     with patch.object(sys, "exit") as mock_exit:
         runpy.run_path(bandscope_analysis.youtube.__file__, run_name="__main__")
         mock_exit.assert_called_with(0)
diff --git a/services/analysis-engine/tests/test_youtube_downloaded_duration_revalidation.py b/services/analysis-engine/tests/test_youtube_downloaded_duration_revalidation.py
new file mode 100644
index 000000000..6780203f6
--- /dev/null
+++ b/services/analysis-engine/tests/test_youtube_downloaded_duration_revalidation.py
@@ -0,0 +1,40 @@
+"""Post-download YouTube duration revalidation regressions."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bandscope_analysis.youtube import download_youtube_audio
+
+
+@patch("bandscope_analysis.youtube.os.path.exists")
+@patch("bandscope_analysis.youtube.os.path.isfile")
+@patch("bandscope_analysis.youtube.os.remove")
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_youtube_revalidates_downloaded_duration_before_returning_success(
+    mock_ydl_class: MagicMock,
+    mock_remove: MagicMock,
+    mock_isfile: MagicMock,
+    mock_exists: MagicMock,
+) -> None:
+    """Changed download metadata must not bypass the 15-minute admission limit."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    out_dir = str(Path("/tmp").resolve())
+    mock_ydl.extract_info.side_effect = [
+        {"id": "abc123DEF45", "duration": 60},
+        {"id": "abc123DEF45", "title": "Changed metadata", "duration": 16 * 60},
+    ]
+    mock_ydl.prepare_filename.return_value = f"{out_dir}/abc123DEF45.m4a"
+    mock_exists.return_value = True
+    mock_isfile.return_value = True
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", out_dir)
+
+    assert result == {
+        "ok": False,
+        "error": {
+            "code": "duration_exceeded",
+            "message": "Video exceeds the 15-minute limit.",
+        },
+    }
+    mock_remove.assert_called_once_with(f"{out_dir}/abc123DEF45.m4a")
diff --git a/services/analysis-engine/tests/test_youtube_duration_contract.py b/services/analysis-engine/tests/test_youtube_duration_contract.py
new file mode 100644
index 000000000..0cb168787
--- /dev/null
+++ b/services/analysis-engine/tests/test_youtube_duration_contract.py
@@ -0,0 +1,54 @@
+"""Fail-closed YouTube duration metadata admission contract."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from bandscope_analysis.youtube import download_youtube_audio
+
+
+class _NonCanonicalFloat(float):
+    """Numeric subtype that must not cross the untrusted metadata boundary."""
+
+
+@pytest.mark.parametrize(
+    "duration",
+    [
+        True,
+        0,
+        -1,
+        float("nan"),
+        float("inf"),
+        "60",
+        object(),
+        _NonCanonicalFloat(60.0),
+    ],
+)
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_youtube_rejects_malformed_announced_duration_before_download(
+    mock_ydl_class: MagicMock,
+    duration: object,
+) -> None:
+    """Malformed known-duration metadata must not authorize a media download."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {
+        "id": "abc123DEF45",
+        "duration": duration,
+    }
+
+    result = download_youtube_audio("https://youtube.com/watch?v=abc123DEF45", "/tmp")
+
+    assert result == {
+        "ok": False,
+        "error": {
+            "code": "download_error",
+            "message": "YouTube import failed. Please use a local audio file instead.",
+        },
+    }
+    mock_ydl.extract_info.assert_called_once_with(
+        "https://youtube.com/watch?v=abc123DEF45",
+        download=False,
+    )
diff --git a/services/analysis-engine/tests/test_youtube_post_download_path_authority.py b/services/analysis-engine/tests/test_youtube_post_download_path_authority.py
new file mode 100644
index 000000000..75b23c5eb
--- /dev/null
+++ b/services/analysis-engine/tests/test_youtube_post_download_path_authority.py
@@ -0,0 +1,74 @@
+"""Regression coverage for post-download YouTube path authority.
+
+The downloader owns only artifacts that resolve beneath the per-import output
+directory. Metadata returned by yt-dlp must not turn an arbitrary filesystem path
+into a successful import or deletion target.
+"""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bandscope_analysis.audio_resource_policy import DEFAULT_MAX_ENCODED_FILE_BYTES
+from bandscope_analysis.youtube import YOUTUBE_IMPORT_FAILED_MESSAGE, download_youtube_audio
+
+
+def _configure_download(mock_ydl_class: MagicMock, filepath: Path) -> None:
+    """Configure yt-dlp to report one completed download at ``filepath``."""
+    mock_ydl = MagicMock()
+    mock_ydl_class.return_value.__enter__.return_value = mock_ydl
+    mock_ydl.extract_info.return_value = {
+        "id": "abc123DEF45",
+        "title": "Authority regression",
+        "duration": 60,
+    }
+    mock_ydl.prepare_filename.return_value = str(filepath)
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_download_rejects_foreign_completed_path(
+    mock_ydl_class: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """A completed path outside this import directory must never become success metadata."""
+    out_dir = tmp_path / "import-cache"
+    out_dir.mkdir()
+    foreign = tmp_path / "foreign.m4a"
+    foreign.write_bytes(b"not-owned-by-this-import")
+    _configure_download(mock_ydl_class, foreign)
+
+    result = download_youtube_audio(
+        "https://youtube.com/watch?v=abc123DEF45",
+        str(out_dir),
+    )
+
+    assert result == {
+        "ok": False,
+        "error": {"code": "download_error", "message": YOUTUBE_IMPORT_FAILED_MESSAGE},
+    }
+    assert foreign.read_bytes() == b"not-owned-by-this-import"
+
+
+@patch("bandscope_analysis.youtube.yt_dlp.YoutubeDL")
+def test_oversize_foreign_completed_path_is_not_deleted(
+    mock_ydl_class: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """Oversize rejection must not delete a path outside this import's authority."""
+    out_dir = tmp_path / "import-cache"
+    out_dir.mkdir()
+    foreign = tmp_path / "foreign-oversize.m4a"
+    with foreign.open("wb") as handle:
+        handle.truncate(DEFAULT_MAX_ENCODED_FILE_BYTES + 1)
+    _configure_download(mock_ydl_class, foreign)
+
+    result = download_youtube_audio(
+        "https://youtube.com/watch?v=abc123DEF45",
+        str(out_dir),
+    )
+
+    assert result == {
+        "ok": False,
+        "error": {"code": "download_error", "message": YOUTUBE_IMPORT_FAILED_MESSAGE},
+    }
+    assert foreign.exists()
+    assert foreign.stat().st_size == DEFAULT_MAX_ENCODED_FILE_BYTES + 1