diff --git a/crates/silo-cli/src/main.rs b/crates/silo-cli/src/main.rs
index bf841eb..84820bb 100644
--- a/crates/silo-cli/src/main.rs
+++ b/crates/silo-cli/src/main.rs
@@ -286,6 +286,17 @@ enum RepoCommand {
         basic_auth: Option<String>,
         #[arg(long, conflicts_with = "basic_auth")]
         bearer_token: Option<String>,
+        /// Tried highest first; ties fall back to the upstream name.
+        #[arg(long, default_value_t = 0)]
+        priority: i32,
+        /// Only consult this upstream for package names matching this
+        /// glob (`*`, `?`). Repeat for several — a name matching any one
+        /// of them is served. Without any, the upstream answers for every
+        /// name. Use it when two upstreams of one format are not
+        /// interchangeable mirrors, e.g. `--package-pattern '@acme/*'`
+        /// for a vendor registry sitting next to a public one.
+        #[arg(long = "package-pattern")]
+        package_patterns: Vec<String>,
         /// apk/pacman/deb: which architectures to sync. Repeat for several.
         #[arg(long = "arch")]
         arches: Vec<String>,
@@ -335,6 +346,17 @@ enum RepoCommand {
         /// Same as `--clear-arches`, for the component list.
         #[arg(long, conflicts_with = "components")]
         clear_components: bool,
+        /// Tried highest first; ties fall back to the upstream name.
+        #[arg(long)]
+        priority: Option<i32>,
+        /// Replaces the package-name globs this upstream answers for.
+        /// Repeat for several. See `add-upstream`.
+        #[arg(long = "package-pattern")]
+        package_patterns: Vec<String>,
+        /// Same as `--clear-arches`, for the package patterns — leaves
+        /// the upstream answering for every name again.
+        #[arg(long, conflicts_with = "package_patterns")]
+        clear_package_patterns: bool,
     },
     /// Removes an upstream. Cached packages it produced keep serving,
     /// relabeled as local, unless `--prune` is given. Admin only.
@@ -1344,6 +1366,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
             arches,
             suite,
             components,
+            priority,
+            package_patterns,
         } => {
             if !cache && !no_cache {
                 anyhow::bail!("one of --cache or --no-cache is required");
@@ -1359,6 +1383,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
             let response = client
                 .add_upstream(with_auth(
                     AddUpstreamRequest {
+                        priority,
+                        package_patterns,
                         repo,
                         channel,
                         name,
@@ -1394,6 +1420,9 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
             components,
             clear_arches,
             clear_components,
+            priority,
+            package_patterns,
+            clear_package_patterns,
         } => {
             let cache_mode = match (cache, no_cache) {
                 (true, false) => Some(UpstreamCacheMode::Cache as i32),
@@ -1418,6 +1447,9 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
             let response = client
                 .update_upstream(with_auth(
                     UpdateUpstreamRequest {
+                        priority,
+                        package_patterns,
+                        clear_package_patterns,
                         repo,
                         channel,
                         name,
@@ -1484,6 +1516,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
                             "base_url": u.base_url,
                             "cache_mode": upstream_cache_mode_name(u.cache_mode),
                             "cache_index_in_memory": u.cache_index_in_memory,
+                            "priority": u.priority,
+                            "package_patterns": u.package_patterns,
                             "auth_configured": u.auth_configured,
                             "status": u.status,
                             "last_sync_at": u.last_sync_at,
@@ -1493,11 +1527,16 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
                     .collect();
                 return print_json(&json!(upstreams));
             }
+            // Listed in the order they are actually tried, with the two
+            // things that decide it — a repo whose upstreams are not
+            // interchangeable mirrors is unreadable without them.
             let mut table = Table::new(&[
                 "NAME",
                 "FORMAT",
                 "BASE_URL",
                 "CACHE",
+                "PRIO",
+                "PACKAGES",
                 "AUTH",
                 "STATUS",
                 "LAST_SYNC",
@@ -1508,6 +1547,12 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) ->
                     format_name(u.format),
                     u.base_url.clone(),
                     upstream_cache_mode_name(u.cache_mode).to_string(),
+                    u.priority.to_string(),
+                    if u.package_patterns.is_empty() {
+                        "*".to_string()
+                    } else {
+                        u.package_patterns.join(",")
+                    },
                     u.auth_configured.to_string(),
                     u.status.clone(),
                     timestamp(u.last_sync_at),
@@ -1582,6 +1627,15 @@ fn print_upstream(upstream: Option<&UpstreamInfo>) {
     println!("  base_url:    {}", u.base_url);
     println!("  cache_mode:  {}", upstream_cache_mode_name(u.cache_mode));
     println!("  index cache: {}", u.cache_index_in_memory);
+    println!("  priority:    {}", u.priority);
+    println!(
+        "  packages:    {}",
+        if u.package_patterns.is_empty() {
+            "* (every name)".to_string()
+        } else {
+            u.package_patterns.join(", ")
+        }
+    );
     println!(
         "  auth:        {}",
         if u.auth_configured { "configured" } else { "-" }
diff --git a/crates/silo-core/src/pull_through.rs b/crates/silo-core/src/pull_through.rs
index 41120c3..d66d880 100644
--- a/crates/silo-core/src/pull_through.rs
+++ b/crates/silo-core/src/pull_through.rs
@@ -141,13 +141,16 @@ fn fetch_action(cache_mode: CacheMode, upstream_requires_auth: bool) -> Action {
 }
 
 /// Lists which configured upstreams back a `(repo, channel, format)`
-/// triple, in the order they should be tried. Multiple upstreams of the
-/// same format may be configured (the user's "one or more" ask); they're
-/// tried in name order — a deterministic, operator-controllable order
-/// (rename to reorder) without a separate priority column to manage.
-/// Callers must fall through to the next candidate on a confirmed miss
-/// rather than stopping at the first one, or every upstream after the
-/// first is unreachable in practice.
+/// triple, in the order they should be tried: highest `priority` first,
+/// ties broken by name so the order is always total and stable. Callers
+/// must fall through to the next candidate on a confirmed miss rather
+/// than stopping at the first one, or every upstream after the first is
+/// unreachable in practice.
+///
+/// This does not apply [`upstream_serves`] — the package name isn't
+/// always known this early (an rpm request carries a filename, not a
+/// name). Callers that do know it filter with `upstream_serves` as they
+/// go.
 pub async fn select_upstreams(
     db: &Db,
     repo: &str,
@@ -159,6 +162,64 @@ pub async fn select_upstreams(
     Ok(upstreams)
 }
 
+/// Whether `upstream` is allowed to answer for `package_name`.
+///
+/// An upstream with no patterns answers for everything, which is what
+/// every upstream does until someone says otherwise. Patterns exist for
+/// the case where two upstreams of one format are *not* interchangeable
+/// mirrors: a vendor registry that holds one scope, next to a public one
+/// that holds the rest. Without them the only thing deciding which
+/// upstream serves a name is which one answers first — and an upstream
+/// that proxies or redirects unknown names to the public registry
+/// answers for everything, so it wins everything, and every package in
+/// the repo ends up attributed to it.
+pub fn upstream_serves(upstream: &UpstreamRow, package_name: &str) -> bool {
+    upstream.package_patterns.is_empty()
+        || upstream
+            .package_patterns
+            .iter()
+            .any(|pattern| glob_matches(pattern, package_name))
+}
+
+/// Matches `name` against a glob: `*` stands for any run of characters
+/// (including none, and including `/`), `?` for exactly one, and
+/// everything else is literal. The whole name must match, not a prefix —
+/// `@acme/*` is a scope, not "anything starting with `@acme/`" that
+/// could also match `@acme/x/../y`.
+///
+/// `*` deliberately spans `/` so `@acme/*` covers the whole scope. npm
+/// names have at most one separator and the other formats have none, so
+/// there is no nesting for a stricter `*` to protect.
+///
+/// Hand-rolled rather than pulled in: this is the entire feature, the
+/// two metacharacters are the two an operator expects from a shell glob,
+/// and backtracking on inputs this short is free.
+fn glob_matches(pattern: &str, name: &str) -> bool {
+    let pattern: Vec<char> = pattern.chars().collect();
+    let name: Vec<char> = name.chars().collect();
+    // `star` remembers the last `*` and how much of `name` it had eaten,
+    // so a failed match resumes by letting that `*` swallow one more
+    // character instead of giving up.
+    let (mut p, mut n) = (0usize, 0usize);
+    let mut star: Option<(usize, usize)> = None;
+    while n < name.len() {
+        if p < pattern.len() && (pattern[p] == '?' || pattern[p] == name[n]) {
+            p += 1;
+            n += 1;
+        } else if p < pattern.len() && pattern[p] == '*' {
+            star = Some((p, n));
+            p += 1;
+        } else if let Some((star_p, star_n)) = star {
+            p = star_p + 1;
+            n = star_n + 1;
+            star = Some((star_p, star_n + 1));
+        } else {
+            return false;
+        }
+    }
+    pattern[p..].iter().all(|c| *c == '*')
+}
+
 /// Whether decrypting `upstream`'s stored credential (if any) would
 /// succeed — used to decide `ProxyUpstream` vs `RedirectToUpstream`
 /// without actually needing the plaintext at this point.
@@ -280,6 +341,100 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_glob_matches_the_whole_name_not_a_prefix() {
+        assert!(glob_matches("lodash", "lodash"));
+        assert!(!glob_matches("lodash", "lodash-es"));
+        assert!(!glob_matches("odash", "lodash"));
+    }
+
+    #[test]
+    fn a_scope_glob_covers_the_scope_and_nothing_else() {
+        assert!(glob_matches(
+            "@fortawesome/*",
+            "@fortawesome/fontawesome-pro"
+        ));
+        assert!(glob_matches(
+            "@fortawesome/*",
+            "@fortawesome/vue-fontawesome"
+        ));
+        // The one this feature exists for: a public package must not fall
+        // inside a vendor registry's scope.
+        assert!(!glob_matches("@fortawesome/*", "@babel/core"));
+        assert!(!glob_matches("@fortawesome/*", "lodash"));
+        // `*` spans `/`, so a scope glob covers the whole scope, but the
+        // scope prefix itself still has to match exactly.
+        assert!(!glob_matches("@fortawesome/*", "@fortawesomeX/thing"));
+    }
+
+    #[test]
+    fn a_star_matches_an_empty_run_and_a_question_mark_matches_exactly_one() {
+        assert!(glob_matches("*", ""));
+        assert!(glob_matches("*", "anything/at-all"));
+        assert!(glob_matches("@acme/*", "@acme/"));
+        assert!(glob_matches("nod?", "node"));
+        assert!(!glob_matches("nod?", "nod"));
+        assert!(!glob_matches("nod?", "nodejs"));
+    }
+
+    #[test]
+    fn several_stars_and_trailing_literals_still_match() {
+        assert!(glob_matches("*-plugin-*", "eslint-plugin-vue"));
+        assert!(glob_matches("@*/core", "@babel/core"));
+        assert!(!glob_matches("@*/core", "@babel/parser"));
+        assert!(glob_matches("**", "anything"));
+    }
+
+    #[test]
+    fn an_upstream_without_patterns_answers_for_everything() {
+        let mut upstream = test_upstream(vec![]);
+        assert!(upstream_serves(&upstream, "anything"));
+        upstream.package_patterns = vec!["@acme/*".into()];
+        assert!(upstream_serves(&upstream, "@acme/widget"));
+        assert!(!upstream_serves(&upstream, "lodash"));
+    }
+
+    #[test]
+    fn patterns_are_alternatives_so_one_upstream_can_hold_several_scopes() {
+        let upstream = test_upstream(vec![
+            "@acme/*".into(),
+            "@vendor/*".into(),
+            "legacy-tool".into(),
+        ]);
+        assert!(upstream_serves(&upstream, "@acme/widget"));
+        assert!(upstream_serves(&upstream, "@vendor/thing"));
+        assert!(upstream_serves(&upstream, "legacy-tool"));
+        assert!(!upstream_serves(&upstream, "@other/thing"));
+    }
+
+    fn test_upstream(package_patterns: Vec<String>) -> UpstreamRow {
+        UpstreamRow {
+            id: silo_db::Uuid::nil(),
+            repo: "r".into(),
+            channel: "c".into(),
+            name: "n".into(),
+            format: "npm".into(),
+            base_url: "https://example.com".into(),
+            cache_mode: "cache".into(),
+            cache_index_in_memory: false,
+            priority: 0,
+            package_patterns,
+            arches: vec![],
+            suite: None,
+            components: vec![],
+            auth_kind: None,
+            auth_username: None,
+            auth_secret_ciphertext: None,
+            auth_secret_nonce: None,
+            status: "ok".into(),
+            last_sync_at: None,
+            last_sync_error: None,
+            last_success_at: None,
+            created_at: chrono::DateTime::from_timestamp(0, 0).unwrap(),
+            updated_at: chrono::DateTime::from_timestamp(0, 0).unwrap(),
+        }
+    }
+
     #[test]
     fn cache_mode_parses_the_two_stored_strings_and_rejects_anything_else() {
         assert_eq!(CacheMode::parse("cache").unwrap(), CacheMode::Cache);
diff --git a/crates/silo-core/src/repo.rs b/crates/silo-core/src/repo.rs
index d49518e..d561761 100644
--- a/crates/silo-core/src/repo.rs
+++ b/crates/silo-core/src/repo.rs
@@ -507,6 +507,13 @@ async fn merge_upstream_records(
             // own group is regenerated; a synthetic upstream row needs
             // the same check applied explicitly, since it never goes
             // through that regeneration path itself.
+            // An upstream restricted to a set of package names must not
+            // advertise anything else, or a client would ask for a name
+            // this upstream is not supposed to serve and the miss path
+            // would have to turn it away after the fact.
+            if !crate::pull_through::upstream_serves(upstream, &row.name) {
+                continue;
+            }
             let belongs = match format {
                 PackageFormat::Apk | PackageFormat::Pacman => {
                     row.arch == index_group
diff --git a/crates/silo-core/src/upstream_sync.rs b/crates/silo-core/src/upstream_sync.rs
index 6e219a6..39f1e81 100644
--- a/crates/silo-core/src/upstream_sync.rs
+++ b/crates/silo-core/src/upstream_sync.rs
@@ -253,6 +253,8 @@ mod tests {
             base_url: "https://example.com/repo".into(),
             cache_mode: "cache".into(),
             cache_index_in_memory: false,
+            priority: 0,
+            package_patterns: vec![],
             arches: vec![],
             suite: None,
             components: vec![],
diff --git a/crates/silo-db/migrations/0006_upstream_routing.sql b/crates/silo-db/migrations/0006_upstream_routing.sql
new file mode 100644
index 0000000..0d829ac
--- /dev/null
+++ b/crates/silo-db/migrations/0006_upstream_routing.sql
@@ -0,0 +1,19 @@
+-- Which upstream answers for a given package, when a repo/channel has
+-- more than one of the same format.
+--
+-- Without these two columns the only ordering is the name, alphabetically,
+-- and every upstream is asked about every package. That is wrong the
+-- moment the upstreams are not interchangeable mirrors — a vendor registry
+-- holding one scope alongside a public one, say. It also fails quietly
+-- rather than loudly: an upstream that answers for a name it has no
+-- business serving (proxying it, or redirecting to the public registry)
+-- simply wins, and every package ends up attributed to it.
+
+-- Higher is tried first; ties fall back to the name, so an existing
+-- deployment that has never set a priority keeps the order it has today.
+ALTER TABLE upstreams ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;
+
+-- Globs the package name must match for this upstream to be consulted at
+-- all. Empty means "no restriction", which is what every existing row
+-- gets, so adding this changes nothing until an operator opts in.
+ALTER TABLE upstreams ADD COLUMN package_patterns TEXT[] NOT NULL DEFAULT '{}';
diff --git a/crates/silo-db/src/upstreams.rs b/crates/silo-db/src/upstreams.rs
index 65361da..38f345d 100644
--- a/crates/silo-db/src/upstreams.rs
+++ b/crates/silo-db/src/upstreams.rs
@@ -22,6 +22,12 @@ pub struct UpstreamRow {
     pub base_url: String,
     pub cache_mode: String,
     pub cache_index_in_memory: bool,
+    /// Tried highest first. See `silo_core::pull_through::select_upstreams`.
+    pub priority: i32,
+    /// Globs the package name must match for this upstream to be
+    /// consulted. Empty means no restriction. See
+    /// `silo_core::pull_through::upstream_serves`.
+    pub package_patterns: Vec<String>,
     pub arches: Vec<String>,
     pub suite: Option<String>,
     pub components: Vec<String>,
@@ -57,16 +63,34 @@ pub struct NewUpstream {
     pub base_url: String,
     pub cache_mode: String,
     pub cache_index_in_memory: bool,
+    pub priority: i32,
+    pub package_patterns: Vec<String>,
     pub arches: Vec<String>,
     pub suite: Option<String>,
     pub components: Vec<String>,
     pub auth: Option<SealedAuth>,
 }
 
+/// The mutable half of an upstream: everything `update-upstream` can
+/// change, as one value rather than a positional argument list nobody can
+/// read at the call site.
+#[derive(Debug, Clone)]
+pub struct UpstreamSettings<'a> {
+    pub base_url: &'a str,
+    pub cache_mode: &'a str,
+    pub cache_index_in_memory: bool,
+    pub priority: i32,
+    pub package_patterns: &'a [String],
+    pub arches: &'a [String],
+    pub suite: Option<&'a str>,
+    pub components: &'a [String],
+}
+
 const COLUMNS: &str = "id, repo, channel, name, format, base_url, cache_mode, \
-                       cache_index_in_memory, arches, suite, components, auth_kind, \
-                       auth_username, auth_secret_ciphertext, auth_secret_nonce, status, \
-                       last_sync_at, last_sync_error, last_success_at, created_at, updated_at";
+                       cache_index_in_memory, priority, package_patterns, arches, suite, \
+                       components, auth_kind, auth_username, auth_secret_ciphertext, \
+                       auth_secret_nonce, status, last_sync_at, last_sync_error, \
+                       last_success_at, created_at, updated_at";
 
 impl Db {
     /// Inserts a new upstream row. Fails on a `(repo, channel, name)`
@@ -85,10 +109,10 @@ impl Db {
         };
         Ok(sqlx::query_as(&format!(
             "INSERT INTO upstreams (repo, channel, name, format, base_url, cache_mode, \
-                                    cache_index_in_memory, arches, suite, components, \
-                                    auth_kind, auth_username, auth_secret_ciphertext, \
-                                    auth_secret_nonce, status) \
-             VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,'ok') \
+                                    cache_index_in_memory, priority, package_patterns, \
+                                    arches, suite, components, auth_kind, auth_username, \
+                                    auth_secret_ciphertext, auth_secret_nonce, status) \
+             VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,'ok') \
              RETURNING {COLUMNS}"
         ))
         .bind(&new.repo)
@@ -98,6 +122,8 @@ impl Db {
         .bind(&new.base_url)
         .bind(&new.cache_mode)
         .bind(new.cache_index_in_memory)
+        .bind(new.priority)
+        .bind(&new.package_patterns)
         .bind(&new.arches)
         .bind(&new.suite)
         .bind(&new.components)
@@ -171,33 +197,30 @@ impl Db {
     /// stored credential untouched; clearing it is a distinct explicit
     /// action (`clear_upstream_auth`) so a caller can't accidentally wipe
     /// a working credential by omitting `auth` from an unrelated update.
-    #[allow(clippy::too_many_arguments)]
     pub async fn update_upstream(
         &self,
         id: Uuid,
-        base_url: &str,
-        cache_mode: &str,
-        cache_index_in_memory: bool,
-        arches: &[String],
-        suite: Option<&str>,
-        components: &[String],
+        settings: &UpstreamSettings<'_>,
         auth: Option<&SealedAuth>,
     ) -> anyhow::Result<Option<UpstreamRow>> {
         if let Some(auth) = auth {
             return Ok(sqlx::query_as(&format!(
                 "UPDATE upstreams SET base_url = $2, cache_mode = $3, \
-                     cache_index_in_memory = $4, arches = $5, suite = $6, components = $7, \
-                     auth_kind = $8, auth_username = $9, auth_secret_ciphertext = $10, \
-                     auth_secret_nonce = $11, updated_at = now() \
+                     cache_index_in_memory = $4, priority = $5, package_patterns = $6, \
+                     arches = $7, suite = $8, components = $9, \
+                     auth_kind = $10, auth_username = $11, auth_secret_ciphertext = $12, \
+                     auth_secret_nonce = $13, updated_at = now() \
                  WHERE id = $1 RETURNING {COLUMNS}"
             ))
             .bind(id)
-            .bind(base_url)
-            .bind(cache_mode)
-            .bind(cache_index_in_memory)
-            .bind(arches)
-            .bind(suite)
-            .bind(components)
+            .bind(settings.base_url)
+            .bind(settings.cache_mode)
+            .bind(settings.cache_index_in_memory)
+            .bind(settings.priority)
+            .bind(settings.package_patterns)
+            .bind(settings.arches)
+            .bind(settings.suite)
+            .bind(settings.components)
             .bind(&auth.kind)
             .bind(&auth.username)
             .bind(&auth.ciphertext)
@@ -207,25 +230,23 @@ impl Db {
         }
         Ok(sqlx::query_as(&format!(
             "UPDATE upstreams SET base_url = $2, cache_mode = $3, \
-                 cache_index_in_memory = $4, arches = $5, suite = $6, components = $7, \
-                 updated_at = now() \
+                 cache_index_in_memory = $4, priority = $5, package_patterns = $6, \
+                 arches = $7, suite = $8, components = $9, updated_at = now() \
              WHERE id = $1 RETURNING {COLUMNS}"
         ))
         .bind(id)
-        .bind(base_url)
-        .bind(cache_mode)
-        .bind(cache_index_in_memory)
-        .bind(arches)
-        .bind(suite)
-        .bind(components)
+        .bind(settings.base_url)
+        .bind(settings.cache_mode)
+        .bind(settings.cache_index_in_memory)
+        .bind(settings.priority)
+        .bind(settings.package_patterns)
+        .bind(settings.arches)
+        .bind(settings.suite)
+        .bind(settings.components)
         .fetch_optional(self.pool())
         .await?)
     }
 
-    /// Clears a stored credential without touching anything else —
-    /// `update_upstream` only ever *sets* one, so a caller that wants to
-    /// remove a credential entirely (rather than replace it) uses this
-    /// instead.
     pub async fn clear_upstream_auth(&self, id: Uuid) -> anyhow::Result<()> {
         sqlx::query(
             "UPDATE upstreams SET auth_kind = NULL, auth_username = NULL, \
@@ -558,7 +579,8 @@ pub async fn list_in_channel<'e, E: PgExecutor<'e>>(
     channel: &str,
 ) -> anyhow::Result<Vec<UpstreamRow>> {
     Ok(sqlx::query_as(&format!(
-        "SELECT {COLUMNS} FROM upstreams WHERE repo = $1 AND channel = $2 ORDER BY name"
+        "SELECT {COLUMNS} FROM upstreams WHERE repo = $1 AND channel = $2 \
+         ORDER BY priority DESC, name"
     ))
     .bind(repo)
     .bind(channel)
@@ -633,6 +655,8 @@ mod tests {
             base_url: "https://example.com/repo".to_string(),
             cache_mode: "cache".to_string(),
             cache_index_in_memory: false,
+            priority: 0,
+            package_patterns: vec![],
             arches: vec![],
             suite: None,
             components: vec![],
@@ -681,6 +705,73 @@ mod tests {
         assert_eq!(listed[0].name, "epel");
     }
 
+    /// Upstreams come back in the order they should be tried: highest
+    /// priority first, and the name only as a tie-break. Ordering by name
+    /// alone means the only way to reorder is to rename, and which
+    /// upstream serves a package is then an accident of the alphabet.
+    #[tokio::test]
+    async fn upstreams_are_listed_highest_priority_first_then_by_name() {
+        let Some(db) = db().await else {
+            eprintln!("skipping: set SILO_TEST_DATABASE_URL");
+            return;
+        };
+        let repo = unique("prio");
+        for (name, priority) in [("aaa", 0), ("mmm", 10), ("zzz", 10), ("bbb", -5)] {
+            let mut new = new_upstream(&repo, "stable", name);
+            new.priority = priority;
+            db.create_upstream(&new).await.unwrap();
+        }
+
+        let listed = db.list_upstreams(&repo, "stable").await.unwrap();
+        let order: Vec<&str> = listed.iter().map(|u| u.name.as_str()).collect();
+        assert_eq!(order, vec!["mmm", "zzz", "aaa", "bbb"]);
+    }
+
+    /// Both routing columns round-trip, and an upstream that has never
+    /// been given either behaves exactly as it did before they existed.
+    #[tokio::test]
+    async fn routing_settings_round_trip_and_default_to_unrestricted() {
+        let Some(db) = db().await else {
+            eprintln!("skipping: set SILO_TEST_DATABASE_URL");
+            return;
+        };
+        let repo = unique("routing");
+        let plain = db
+            .create_upstream(&new_upstream(&repo, "stable", "plain"))
+            .await
+            .unwrap();
+        assert_eq!(plain.priority, 0);
+        assert!(plain.package_patterns.is_empty());
+
+        let mut scoped = new_upstream(&repo, "stable", "scoped");
+        scoped.priority = 7;
+        scoped.package_patterns = vec!["@acme/*".into(), "legacy-tool".into()];
+        let scoped = db.create_upstream(&scoped).await.unwrap();
+        assert_eq!(scoped.priority, 7);
+        assert_eq!(scoped.package_patterns, vec!["@acme/*", "legacy-tool"]);
+
+        let updated = db
+            .update_upstream(
+                scoped.id,
+                &UpstreamSettings {
+                    base_url: &scoped.base_url,
+                    cache_mode: &scoped.cache_mode,
+                    cache_index_in_memory: scoped.cache_index_in_memory,
+                    priority: -3,
+                    package_patterns: &[],
+                    arches: &scoped.arches,
+                    suite: scoped.suite.as_deref(),
+                    components: &scoped.components,
+                },
+                None,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(updated.priority, -3);
+        assert!(updated.package_patterns.is_empty());
+    }
+
     #[tokio::test]
     async fn duplicate_name_within_repo_channel_is_rejected() {
         let Some(db) = db().await else {
diff --git a/crates/silo-server/src/admin.rs b/crates/silo-server/src/admin.rs
index 6883bc7..e0f4ea4 100644
--- a/crates/silo-server/src/admin.rs
+++ b/crates/silo-server/src/admin.rs
@@ -1265,6 +1265,8 @@ impl AdminServiceImpl {
             ));
         }
 
+        validate_package_patterns(&req.package_patterns)?;
+
         let opts = silo_pkg::UpstreamFetchOptions {
             arches: req.arches.clone(),
             suite: non_empty_str(&req.suite).map(str::to_string),
@@ -1306,6 +1308,8 @@ impl AdminServiceImpl {
                 base_url: req.base_url.clone(),
                 cache_mode: cache_mode.to_string(),
                 cache_index_in_memory: req.cache_index_in_memory,
+                priority: req.priority,
+                package_patterns: req.package_patterns.clone(),
                 arches: req.arches.clone(),
                 suite: opts.suite.clone(),
                 components: req.components.clone(),
@@ -1421,6 +1425,25 @@ impl AdminServiceImpl {
         } else {
             req.components.clone()
         };
+        let priority = req.priority.unwrap_or(existing.priority);
+        let package_patterns = if req.clear_package_patterns {
+            Vec::new()
+        } else if req.package_patterns.is_empty() {
+            existing.package_patterns.clone()
+        } else {
+            req.package_patterns.clone()
+        };
+        validate_package_patterns(&package_patterns)?;
+        let settings = silo_db::upstreams::UpstreamSettings {
+            base_url: &base_url,
+            cache_mode: &cache_mode,
+            cache_index_in_memory,
+            priority,
+            package_patterns: &package_patterns,
+            arches: &arches,
+            suite: suite.as_deref(),
+            components: &components,
+        };
 
         let sealed = match &req.auth {
             None => None,
@@ -1446,16 +1469,7 @@ impl AdminServiceImpl {
             None => {
                 self.state
                     .db
-                    .update_upstream(
-                        existing.id,
-                        &base_url,
-                        &cache_mode,
-                        cache_index_in_memory,
-                        &arches,
-                        suite.as_deref(),
-                        &components,
-                        None,
-                    )
+                    .update_upstream(existing.id, &settings, None)
                     .await
             }
             Some(None) => {
@@ -1470,31 +1484,13 @@ impl AdminServiceImpl {
                     .map_err(|e| Status::internal(e.to_string()))?;
                 self.state
                     .db
-                    .update_upstream(
-                        existing.id,
-                        &base_url,
-                        &cache_mode,
-                        cache_index_in_memory,
-                        &arches,
-                        suite.as_deref(),
-                        &components,
-                        None,
-                    )
+                    .update_upstream(existing.id, &settings, None)
                     .await
             }
             Some(Some(sealed)) => {
                 self.state
                     .db
-                    .update_upstream(
-                        existing.id,
-                        &base_url,
-                        &cache_mode,
-                        cache_index_in_memory,
-                        &arches,
-                        suite.as_deref(),
-                        &components,
-                        Some(&sealed),
-                    )
+                    .update_upstream(existing.id, &settings, Some(&sealed))
                     .await
             }
         }
@@ -1736,6 +1732,26 @@ fn non_empty_str(s: &str) -> Option<&str> {
 /// would round-trip in plain text through `download_url` and, on a
 /// `no_cache` upstream, straight into a client-visible redirect
 /// `Location` header.
+/// A pattern that can never match anything is always a mistake, and one
+/// that silently routes every package to the wrong upstream — so it is
+/// rejected at the point it is set rather than discovered later from the
+/// `origin` column.
+fn validate_package_patterns(patterns: &[String]) -> Result<(), Status> {
+    for pattern in patterns {
+        if pattern.trim().is_empty() {
+            return Err(Status::invalid_argument(
+                "a package pattern must not be empty",
+            ));
+        }
+        if pattern.trim() != pattern {
+            return Err(Status::invalid_argument(format!(
+                "package pattern `{pattern}` has leading or trailing whitespace",
+            )));
+        }
+    }
+    Ok(())
+}
+
 fn validate_base_url(url: &str) -> Result<(), Status> {
     if !url.starts_with("http://") && !url.starts_with("https://") {
         return Err(Status::invalid_argument(
@@ -1794,6 +1810,8 @@ fn to_proto_upstream(row: &silo_db::upstreams::UpstreamRow) -> UpstreamInfo {
         base_url: row.base_url.clone(),
         cache_mode: cache_mode as i32,
         cache_index_in_memory: row.cache_index_in_memory,
+        priority: row.priority,
+        package_patterns: row.package_patterns.clone(),
         arches: row.arches.clone(),
         suite: row.suite.clone().unwrap_or_default(),
         components: row.components.clone(),
diff --git a/crates/silo-server/src/http.rs b/crates/silo-server/src/http.rs
index b9ae502..8809cc0 100644
--- a/crates/silo-server/src/http.rs
+++ b/crates/silo-server/src/http.rs
@@ -490,6 +490,13 @@ async fn pull_through_miss(
     let mut upstream_pkg = None;
     let mut matched_upstream = None;
     for upstream in upstreams {
+        // An upstream restricted to a set of package names is not asked
+        // about anything else, however eagerly it would answer.
+        if let Some(name) = package_name {
+            if !silo_core::pull_through::upstream_serves(upstream, name) {
+                continue;
+            }
+        }
         // When the in-memory cache is enabled for this upstream, read
         // through it — the same one the index merge uses, so the two
         // lookup paths never disagree about what's fresh. Otherwise this
@@ -750,6 +757,9 @@ async fn lazy_sync_npm_upstream_packages(
     let mut fetched = Vec::new();
     let mut saw_transient_error = false;
     for upstream in upstreams {
+        if !silo_core::pull_through::upstream_serves(upstream, name) {
+            continue;
+        }
         match silo_core::upstream_sync::sync_npm_package(
             &state.db,
             state.upstream_http.clone(),
diff --git a/crates/silo-server/tests/publish_flow.rs b/crates/silo-server/tests/publish_flow.rs
index 4034ca1..6313a78 100644
--- a/crates/silo-server/tests/publish_flow.rs
+++ b/crates/silo-server/tests/publish_flow.rs
@@ -1118,6 +1118,8 @@ async fn a_publish_holds_only_one_pooled_connection_at_a_time() {
             base_url: "https://registry.example".into(),
             cache_mode: "cache".into(),
             cache_index_in_memory: false,
+            priority: 0,
+            package_patterns: vec![],
             arches: vec![],
             suite: None,
             components: vec![],
diff --git a/crates/silo-server/tests/upstream_pull_through.rs b/crates/silo-server/tests/upstream_pull_through.rs
index 8ebaaa9..f12991b 100644
--- a/crates/silo-server/tests/upstream_pull_through.rs
+++ b/crates/silo-server/tests/upstream_pull_through.rs
@@ -98,6 +98,8 @@ async fn add_upstream_validates_and_syncs_before_creating_the_row() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -144,6 +146,8 @@ async fn add_upstream_against_an_unreachable_url_creates_no_row() {
                 base_url: "http://127.0.0.1:1".into(), // nothing listens here
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -206,6 +210,8 @@ async fn setup_cache_upstream(
                 base_url: mock.uri(),
                 cache_mode: cache_mode as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -624,6 +630,8 @@ async fn a_noarch_upstream_package_is_merged_into_every_concrete_architecture()
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["aarch64".into()],
                 suite: String::new(),
@@ -709,6 +717,8 @@ async fn cache_index_in_memory_does_not_change_pull_through_behavior() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: true,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -840,6 +850,8 @@ async fn npm_packument_miss_lazily_syncs_and_renders_from_the_upstream() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -973,6 +985,8 @@ async fn a_cache_mode_signed_rpm_upstream_is_not_advertised_until_actually_fetch
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1093,6 +1107,8 @@ async fn a_second_upstream_is_still_reachable_when_an_earlier_one_by_name_does_n
                 base_url: alpha_mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -1133,6 +1149,8 @@ async fn a_second_upstream_is_still_reachable_when_an_earlier_one_by_name_does_n
                 base_url: zeta_mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec!["x86_64".into()],
                 suite: String::new(),
@@ -1199,6 +1217,8 @@ async fn npm_packument_miss_falls_through_to_a_second_upstream_by_name() {
                 base_url: alpha_mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1246,6 +1266,8 @@ async fn npm_packument_miss_falls_through_to_a_second_upstream_by_name() {
                 base_url: zeta_mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1352,6 +1374,8 @@ async fn a_transient_upstream_failure_surfaces_as_a_retryable_502_not_a_permanen
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1413,6 +1437,8 @@ async fn an_upstream_confirmed_404_still_surfaces_as_an_ordinary_404() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1498,6 +1524,8 @@ async fn a_tarball_request_with_no_prior_packument_fetch_still_lazily_syncs() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1617,6 +1645,8 @@ async fn an_npm_packument_only_merges_its_own_names_upstream_versions() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1720,6 +1750,8 @@ async fn a_tarball_filename_shared_by_two_scoped_packages_resolves_to_the_right_
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1822,6 +1854,8 @@ async fn a_tarball_first_request_resolves_in_one_pass() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: false,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -1935,6 +1969,8 @@ async fn an_npm_lazy_sync_drops_the_upstreams_cached_index() {
                 base_url: mock.uri(),
                 cache_mode: UpstreamCacheMode::Cache as i32,
                 cache_index_in_memory: true,
+                priority: 0,
+                package_patterns: vec![],
                 auth: None,
                 arches: vec![],
                 suite: String::new(),
@@ -2029,6 +2065,8 @@ async fn pull_through_audit_entries_can_be_switched_off_without_silencing_real_p
                     base_url: mock.uri(),
                     cache_mode: UpstreamCacheMode::Cache as i32,
                     cache_index_in_memory: false,
+                    priority: 0,
+                    package_patterns: vec![],
                     auth: None,
                     arches: vec![],
                     suite: String::new(),
@@ -2087,3 +2125,149 @@ async fn pull_through_audit_entries_can_be_switched_off_without_silencing_real_p
         );
     }
 }
+
+/// An upstream restricted to a set of package-name globs is not consulted
+/// for anything else, even when it would happily answer.
+///
+/// This is the shape a vendor registry makes: it holds one scope, sits
+/// next to a public registry that holds the rest, and — like a real one —
+/// answers for public names too by proxying or redirecting to the public
+/// registry. Nothing but a pattern stops it winning every name it is
+/// asked about, and every package in the repo then gets attributed to it.
+#[tokio::test]
+async fn an_upstream_scoped_to_a_pattern_is_not_consulted_for_other_names() {
+    let url = require_db!();
+    let harness = Harness::new(&url).await;
+    let repo = unique_repo("npmscoped");
+    let admin = harness.admin_token().await;
+    let service = AdminServiceImpl {
+        state: harness.state.clone(),
+    };
+
+    // The vendor registry: it has its own scope, and answers for
+    // everything else as well.
+    let vendor = MockServer::start().await;
+    // The public registry.
+    let public = MockServer::start().await;
+    for mock in [&vendor, &public] {
+        Mock::given(method("GET"))
+            .and(path("/"))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .set_body_json(serde_json::json!({ "db_name": "registry" })),
+            )
+            .mount(mock)
+            .await;
+    }
+
+    let packument = |host: &str, name: &str| {
+        serde_json::json!({
+            "name": name,
+            "versions": {
+                "1.0.0": {
+                    "name": name,
+                    "version": "1.0.0",
+                    "dist": {
+                        "tarball": format!("{host}/{name}/-/{}-1.0.0.tgz",
+                            name.rsplit('/').next().unwrap()),
+                        "shasum": "abc123",
+                    },
+                },
+            },
+        })
+    };
+    for name in ["@vendor/widget", "lodash"] {
+        let file = format!("{}-1.0.0.tgz", name.rsplit('/').next().unwrap());
+        Mock::given(method("GET"))
+            .and(path(format!("/{name}")))
+            .respond_with(ResponseTemplate::new(200).set_body_json(packument(&vendor.uri(), name)))
+            .mount(&vendor)
+            .await;
+        Mock::given(method("GET"))
+            .and(path(format!("/{name}/-/{file}")))
+            .respond_with(ResponseTemplate::new(200).set_body_bytes(build_test_npm(name, "1.0.0")))
+            .mount(&vendor)
+            .await;
+    }
+    Mock::given(method("GET"))
+        .and(path("/lodash"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(packument(&public.uri(), "lodash")))
+        .mount(&public)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/lodash/-/lodash-1.0.0.tgz"))
+        .respond_with(ResponseTemplate::new(200).set_body_bytes(build_test_npm("lodash", "1.0.0")))
+        .mount(&public)
+        .await;
+
+    // `vendor` sorts first by name *and* is given the higher priority, so
+    // nothing but the pattern keeps it away from `lodash`.
+    for (name, base_url, patterns, priority) in [
+        ("vendor", vendor.uri(), vec!["@vendor/*".to_string()], 10),
+        ("public", public.uri(), vec![], 0),
+    ] {
+        service
+            .add_upstream(request(
+                AddUpstreamRequest {
+                    repo: repo.clone(),
+                    channel: "stable".into(),
+                    name: name.into(),
+                    format: silo_proto::v1::PackageFormat::Npm as i32,
+                    base_url,
+                    cache_mode: UpstreamCacheMode::Cache as i32,
+                    cache_index_in_memory: false,
+                    priority,
+                    package_patterns: patterns,
+                    auth: None,
+                    arches: vec![],
+                    suite: String::new(),
+                    components: vec![],
+                },
+                &admin.secret,
+            ))
+            .await
+            .expect("add_upstream");
+    }
+    harness.db.set_repo_public(&repo, true).await.unwrap();
+
+    for (package, expected_upstream) in [("@vendor/widget", "vendor"), ("lodash", "public")] {
+        let file = format!("{}-1.0.0.tgz", package.rsplit('/').next().unwrap());
+        let response = get(
+            &harness.state,
+            &format!("/{repo}/stable/npm/{package}/-/{file}"),
+        )
+        .await;
+        assert_eq!(
+            response.status(),
+            axum::http::StatusCode::OK,
+            "GET {package}"
+        );
+
+        let origin = harness
+            .db
+            .list_packages(&repo, "stable", Some(silo_pkg::PackageFormat::Npm))
+            .await
+            .unwrap()
+            .into_iter()
+            .find(|p| p.name == package)
+            .and_then(|p| p.origin_upstream_id)
+            .expect("the package was cached with an origin");
+        let origin = harness.db.find_upstream(origin).await.unwrap().unwrap();
+        assert_eq!(
+            origin.name, expected_upstream,
+            "{package} was served by the wrong upstream"
+        );
+    }
+
+    // The vendor registry was never even asked about the public name.
+    let asked_vendor_for_lodash = vendor
+        .received_requests()
+        .await
+        .unwrap_or_default()
+        .iter()
+        .any(|r| r.url.path() == "/lodash");
+    assert!(
+        !asked_vendor_for_lodash,
+        "a pattern-scoped upstream must not be asked about a name outside its patterns"
+    );
+}
diff --git a/proto/silo/v1/admin.proto b/proto/silo/v1/admin.proto
index cd89c25..16dbe1a 100644
--- a/proto/silo/v1/admin.proto
+++ b/proto/silo/v1/admin.proto
@@ -56,7 +56,8 @@ service AdminService {
   // and creates nothing.
   rpc AddUpstream(AddUpstreamRequest) returns (AddUpstreamResponse);
   // Changes an existing upstream's settings in place (cache mode, index
-  // cache toggle, auth, arches/suite/components). Identity
+  // cache toggle, auth, priority, package patterns, arches/suite/
+  // components). Identity
   // (repo/channel/name/format) can't be changed this way — add a new one
   // instead.
   rpc UpdateUpstream(UpdateUpstreamRequest) returns (UpdateUpstreamResponse);
@@ -409,6 +410,11 @@ message UpstreamInfo {
   int64 last_sync_at = 14;
   string last_sync_error = 15;
   int64 last_success_at = 16;
+  // Tried highest first; ties broken by name.
+  int32 priority = 17;
+  // Globs the package name must match for this upstream to be consulted.
+  // Empty means no restriction.
+  repeated string package_patterns = 18;
 }
 
 message AddUpstreamRequest {
@@ -427,6 +433,14 @@ message AddUpstreamRequest {
   string suite = 10;
   // deb only.
   repeated string components = 11;
+  // Tried highest first; ties broken by name. Defaults to 0, which keeps
+  // the plain by-name order for a channel where nobody sets one.
+  int32 priority = 12;
+  // Globs (`*`, `?`) the package name must match for this upstream to be
+  // consulted at all — `@acme/*` for a vendor registry that holds one
+  // scope next to a public registry that holds the rest. Empty means no
+  // restriction, which is every upstream until someone says otherwise.
+  repeated string package_patterns = 13;
 }
 
 message AddUpstreamResponse {
@@ -454,6 +468,11 @@ message UpdateUpstreamRequest {
   repeated string components = 10;
   bool clear_arches = 11;
   bool clear_components = 12;
+  optional int32 priority = 13;
+  // Same "unset means unchanged" convention as arches/components, with
+  // `clear_package_patterns` to empty the list explicitly.
+  repeated string package_patterns = 14;
+  bool clear_package_patterns = 15;
 }
 
 message UpdateUpstreamResponse {