diff --git a/api/unbounded-storage/config.pb.go b/api/unbounded-storage/config.pb.go
index 188cad640..40f3a1041 100644
--- a/api/unbounded-storage/config.pb.go
+++ b/api/unbounded-storage/config.pb.go
@@ -50,8 +50,11 @@ type Config struct {
 	// finger selection uses weighted rendezvous while successor/predecessor
 	// selection remains purely ring-nearest for routing correctness.
 	TopologyWeighting *TopologyWeighting `protobuf:"bytes,11,opt,name=topology_weighting,json=topologyWeighting,proto3,oneof" json:"topology_weighting,omitempty"`
-	unknownFields     protoimpl.UnknownFields
-	sizeCache         protoimpl.SizeCache
+	// Controls automatic disk selection when disks is empty. Explicit disks are
+	// authoritative and are not filtered by this policy.
+	DiskDiscovery *DiskDiscoveryCfg `protobuf:"bytes,12,opt,name=disk_discovery,json=diskDiscovery,proto3,oneof" json:"disk_discovery,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
 }
 
 func (x *Config) Reset() {
@@ -161,6 +164,13 @@ func (x *Config) GetTopologyWeighting() *TopologyWeighting {
 	return nil
 }
 
+func (x *Config) GetDiskDiscovery() *DiskDiscoveryCfg {
+	if x != nil {
+		return x.DiskDiscovery
+	}
+	return nil
+}
+
 // A precomputed routing table for a single node, produced by a
 // global-view planner. All names reference `PeerSpec.name` values in the
 // `peers` list (each must be present so a fabric connection exists). Ring
@@ -2326,11 +2336,68 @@ func (x *LoadgenFrontendConfig) GetSkipLocalDisk() bool {
 	return false
 }
 
+type DiskDiscoveryCfg struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Exact /dev paths excluded from automatic disk selection. This deny list
+	// does not apply to explicitly configured disks.
+	DeniedPaths []string `protobuf:"bytes,1,rep,name=denied_paths,json=deniedPaths,proto3" json:"denied_paths,omitempty"`
+	// File-backed disk used when automatic discovery finds no eligible devices.
+	// The path defaults to /var/lib/unbounded-storage/cache.disk and the size
+	// defaults to 20 GiB when omitted or partially configured.
+	Fallback      *FileDiskConfig `protobuf:"bytes,2,opt,name=fallback,proto3,oneof" json:"fallback,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *DiskDiscoveryCfg) Reset() {
+	*x = DiskDiscoveryCfg{}
+	mi := &file_config_proto_msgTypes[29]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *DiskDiscoveryCfg) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*DiskDiscoveryCfg) ProtoMessage() {}
+
+func (x *DiskDiscoveryCfg) ProtoReflect() protoreflect.Message {
+	mi := &file_config_proto_msgTypes[29]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use DiskDiscoveryCfg.ProtoReflect.Descriptor instead.
+func (*DiskDiscoveryCfg) Descriptor() ([]byte, []int) {
+	return file_config_proto_rawDescGZIP(), []int{29}
+}
+
+func (x *DiskDiscoveryCfg) GetDeniedPaths() []string {
+	if x != nil {
+		return x.DeniedPaths
+	}
+	return nil
+}
+
+func (x *DiskDiscoveryCfg) GetFallback() *FileDiskConfig {
+	if x != nil {
+		return x.Fallback
+	}
+	return nil
+}
+
 var File_config_proto protoreflect.FileDescriptor
 
 const file_config_proto_rawDesc = "" +
 	"\n" +
-	"\fconfig.proto\x12\x18unbounded.storage.config\"\xcc\x05\n" +
+	"\fconfig.proto\x12\x18unbounded.storage.config\"\xb7\x06\n" +
 	"\x06Config\x12\x18\n" +
 	"\aversion\x18\x01 \x01(\x04R\aversion\x12>\n" +
 	"\astartup\x18\x02 \x01(\v2$.unbounded.storage.config.StartupCfgR\astartup\x12D\n" +
@@ -2343,10 +2410,12 @@ const file_config_proto_rawDesc = "" +
 	"\frouting_plan\x18\t \x01(\v2%.unbounded.storage.config.RoutingPlanH\x01R\vroutingPlan\x88\x01\x01\x128\n" +
 	"\x05peers\x18\n" +
 	" \x03(\v2\".unbounded.storage.config.PeerSpecR\x05peers\x12_\n" +
-	"\x12topology_weighting\x18\v \x01(\v2+.unbounded.storage.config.TopologyWeightingH\x02R\x11topologyWeighting\x88\x01\x01B\x13\n" +
+	"\x12topology_weighting\x18\v \x01(\v2+.unbounded.storage.config.TopologyWeightingH\x02R\x11topologyWeighting\x88\x01\x01\x12V\n" +
+	"\x0edisk_discovery\x18\f \x01(\v2*.unbounded.storage.config.DiskDiscoveryCfgH\x03R\rdiskDiscovery\x88\x01\x01B\x13\n" +
 	"\x11_fingers_per_nodeB\x0f\n" +
 	"\r_routing_planB\x15\n" +
-	"\x13_topology_weighting\"\x8f\x01\n" +
+	"\x13_topology_weightingB\x11\n" +
+	"\x0f_disk_discovery\"\x8f\x01\n" +
 	"\vRoutingPlan\x12\x18\n" +
 	"\afingers\x18\x01 \x03(\tR\afingers\x12!\n" +
 	"\tsuccessor\x18\x02 \x01(\tH\x00R\tsuccessor\x88\x01\x01\x12%\n" +
@@ -2548,7 +2617,11 @@ const file_config_proto_rawDesc = "" +
 	"\x11_keyspace_objectsB\r\n" +
 	"\v_read_bytesB\x14\n" +
 	"\x12_object_size_bytesB\x10\n" +
-	"\x0e_zipf_exponentB@Z>github.com/Azure/unbounded/api/unbounded-storage;storageconfigb\x06proto3"
+	"\x0e_zipf_exponent\"\x8d\x01\n" +
+	"\x10DiskDiscoveryCfg\x12!\n" +
+	"\fdenied_paths\x18\x01 \x03(\tR\vdeniedPaths\x12I\n" +
+	"\bfallback\x18\x02 \x01(\v2(.unbounded.storage.config.FileDiskConfigH\x00R\bfallback\x88\x01\x01B\v\n" +
+	"\t_fallbackB@Z>github.com/Azure/unbounded/api/unbounded-storage;storageconfigb\x06proto3"
 
 var (
 	file_config_proto_rawDescOnce sync.Once
@@ -2562,7 +2635,7 @@ func file_config_proto_rawDescGZIP() []byte {
 	return file_config_proto_rawDescData
 }
 
-var file_config_proto_msgTypes = make([]protoimpl.MessageInfo, 29)
+var file_config_proto_msgTypes = make([]protoimpl.MessageInfo, 30)
 var file_config_proto_goTypes = []any{
 	(*Config)(nil),                // 0: unbounded.storage.config.Config
 	(*RoutingPlan)(nil),           // 1: unbounded.storage.config.RoutingPlan
@@ -2593,6 +2666,7 @@ var file_config_proto_goTypes = []any{
 	(*HttpFrontendConfig)(nil),    // 26: unbounded.storage.config.HttpFrontendConfig
 	(*S3FrontendConfig)(nil),      // 27: unbounded.storage.config.S3FrontendConfig
 	(*LoadgenFrontendConfig)(nil), // 28: unbounded.storage.config.LoadgenFrontendConfig
+	(*DiskDiscoveryCfg)(nil),      // 29: unbounded.storage.config.DiskDiscoveryCfg
 }
 var file_config_proto_depIdxs = []int32{
 	8,  // 0: unbounded.storage.config.Config.startup:type_name -> unbounded.storage.config.StartupCfg
@@ -2603,31 +2677,33 @@ var file_config_proto_depIdxs = []int32{
 	1,  // 5: unbounded.storage.config.Config.routing_plan:type_name -> unbounded.storage.config.RoutingPlan
 	4,  // 6: unbounded.storage.config.Config.peers:type_name -> unbounded.storage.config.PeerSpec
 	2,  // 7: unbounded.storage.config.Config.topology_weighting:type_name -> unbounded.storage.config.TopologyWeighting
-	3,  // 8: unbounded.storage.config.TopologyWeighting.prefix_weights:type_name -> unbounded.storage.config.TopologyPrefixWeight
-	5,  // 9: unbounded.storage.config.PeerSpec.tcp:type_name -> unbounded.storage.config.TcpPeerConfig
-	6,  // 10: unbounded.storage.config.PeerSpec.rdma:type_name -> unbounded.storage.config.RdmaPeerConfig
-	10, // 11: unbounded.storage.config.StartupCfg.memory:type_name -> unbounded.storage.config.MemoryCfg
-	11, // 12: unbounded.storage.config.StartupCfg.fabric:type_name -> unbounded.storage.config.FabricCfg
-	16, // 13: unbounded.storage.config.StartupCfg.topology:type_name -> unbounded.storage.config.TopologyCfg
-	9,  // 14: unbounded.storage.config.StartupCfg.metrics:type_name -> unbounded.storage.config.MetricsCfg
-	12, // 15: unbounded.storage.config.FabricCfg.tcp:type_name -> unbounded.storage.config.TcpFabricBinds
-	13, // 16: unbounded.storage.config.FabricCfg.rdma:type_name -> unbounded.storage.config.RdmaFabricBinds
-	14, // 17: unbounded.storage.config.FabricCfg.auto_rdma:type_name -> unbounded.storage.config.AutoRdmaFabricBinds
-	15, // 18: unbounded.storage.config.RdmaFabricBinds.binds:type_name -> unbounded.storage.config.RdmaFabricBind
-	18, // 19: unbounded.storage.config.DiskSpec.block:type_name -> unbounded.storage.config.BlockDiskConfig
-	19, // 20: unbounded.storage.config.DiskSpec.file:type_name -> unbounded.storage.config.FileDiskConfig
-	21, // 21: unbounded.storage.config.BackendSpec.http:type_name -> unbounded.storage.config.HttpBackendConfig
-	22, // 22: unbounded.storage.config.BackendSpec.s3:type_name -> unbounded.storage.config.S3BackendConfig
-	23, // 23: unbounded.storage.config.BackendSpec.azure:type_name -> unbounded.storage.config.AzureBackendConfig
-	24, // 24: unbounded.storage.config.BackendSpec.fake:type_name -> unbounded.storage.config.FakeBackendConfig
-	26, // 25: unbounded.storage.config.FrontendSpec.http:type_name -> unbounded.storage.config.HttpFrontendConfig
-	27, // 26: unbounded.storage.config.FrontendSpec.s3:type_name -> unbounded.storage.config.S3FrontendConfig
-	28, // 27: unbounded.storage.config.FrontendSpec.loadgen:type_name -> unbounded.storage.config.LoadgenFrontendConfig
-	28, // [28:28] is the sub-list for method output_type
-	28, // [28:28] is the sub-list for method input_type
-	28, // [28:28] is the sub-list for extension type_name
-	28, // [28:28] is the sub-list for extension extendee
-	0,  // [0:28] is the sub-list for field type_name
+	29, // 8: unbounded.storage.config.Config.disk_discovery:type_name -> unbounded.storage.config.DiskDiscoveryCfg
+	3,  // 9: unbounded.storage.config.TopologyWeighting.prefix_weights:type_name -> unbounded.storage.config.TopologyPrefixWeight
+	5,  // 10: unbounded.storage.config.PeerSpec.tcp:type_name -> unbounded.storage.config.TcpPeerConfig
+	6,  // 11: unbounded.storage.config.PeerSpec.rdma:type_name -> unbounded.storage.config.RdmaPeerConfig
+	10, // 12: unbounded.storage.config.StartupCfg.memory:type_name -> unbounded.storage.config.MemoryCfg
+	11, // 13: unbounded.storage.config.StartupCfg.fabric:type_name -> unbounded.storage.config.FabricCfg
+	16, // 14: unbounded.storage.config.StartupCfg.topology:type_name -> unbounded.storage.config.TopologyCfg
+	9,  // 15: unbounded.storage.config.StartupCfg.metrics:type_name -> unbounded.storage.config.MetricsCfg
+	12, // 16: unbounded.storage.config.FabricCfg.tcp:type_name -> unbounded.storage.config.TcpFabricBinds
+	13, // 17: unbounded.storage.config.FabricCfg.rdma:type_name -> unbounded.storage.config.RdmaFabricBinds
+	14, // 18: unbounded.storage.config.FabricCfg.auto_rdma:type_name -> unbounded.storage.config.AutoRdmaFabricBinds
+	15, // 19: unbounded.storage.config.RdmaFabricBinds.binds:type_name -> unbounded.storage.config.RdmaFabricBind
+	18, // 20: unbounded.storage.config.DiskSpec.block:type_name -> unbounded.storage.config.BlockDiskConfig
+	19, // 21: unbounded.storage.config.DiskSpec.file:type_name -> unbounded.storage.config.FileDiskConfig
+	21, // 22: unbounded.storage.config.BackendSpec.http:type_name -> unbounded.storage.config.HttpBackendConfig
+	22, // 23: unbounded.storage.config.BackendSpec.s3:type_name -> unbounded.storage.config.S3BackendConfig
+	23, // 24: unbounded.storage.config.BackendSpec.azure:type_name -> unbounded.storage.config.AzureBackendConfig
+	24, // 25: unbounded.storage.config.BackendSpec.fake:type_name -> unbounded.storage.config.FakeBackendConfig
+	26, // 26: unbounded.storage.config.FrontendSpec.http:type_name -> unbounded.storage.config.HttpFrontendConfig
+	27, // 27: unbounded.storage.config.FrontendSpec.s3:type_name -> unbounded.storage.config.S3FrontendConfig
+	28, // 28: unbounded.storage.config.FrontendSpec.loadgen:type_name -> unbounded.storage.config.LoadgenFrontendConfig
+	19, // 29: unbounded.storage.config.DiskDiscoveryCfg.fallback:type_name -> unbounded.storage.config.FileDiskConfig
+	30, // [30:30] is the sub-list for method output_type
+	30, // [30:30] is the sub-list for method input_type
+	30, // [30:30] is the sub-list for extension type_name
+	30, // [30:30] is the sub-list for extension extendee
+	0,  // [0:30] is the sub-list for field type_name
 }
 
 func init() { file_config_proto_init() }
@@ -2672,13 +2748,14 @@ func file_config_proto_init() {
 	}
 	file_config_proto_msgTypes[26].OneofWrappers = []any{}
 	file_config_proto_msgTypes[28].OneofWrappers = []any{}
+	file_config_proto_msgTypes[29].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_config_proto_rawDesc), len(file_config_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   29,
+			NumMessages:   30,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/api/unbounded-storage/config.proto b/api/unbounded-storage/config.proto
index 5bc743d4f..e40e0e91a 100644
--- a/api/unbounded-storage/config.proto
+++ b/api/unbounded-storage/config.proto
@@ -39,6 +39,10 @@ message Config {
   // finger selection uses weighted rendezvous while successor/predecessor
   // selection remains purely ring-nearest for routing correctness.
   optional TopologyWeighting topology_weighting = 11;
+
+  // Controls automatic disk selection when disks is empty. Explicit disks are
+  // authoritative and are not filtered by this policy.
+  optional DiskDiscoveryCfg disk_discovery = 12;
 }
 
 // A precomputed routing table for a single node, produced by a
@@ -437,3 +441,14 @@ message LoadgenFrontendConfig {
   // measured path focused on remote NVMe + RDMA instead of requester-side reuse.
   bool skip_local_disk = 11;
 }
+
+message DiskDiscoveryCfg {
+  // Exact /dev paths excluded from automatic disk selection. This deny list
+  // does not apply to explicitly configured disks.
+  repeated string denied_paths = 1;
+
+  // File-backed disk used when automatic discovery finds no eligible devices.
+  // The path defaults to /var/lib/unbounded-storage/cache.disk and the size
+  // defaults to 20 GiB when omitted or partially configured.
+  optional FileDiskConfig fallback = 2;
+}
diff --git a/cmd/unbounded-storage/ARCHITECTURE.md b/cmd/unbounded-storage/ARCHITECTURE.md
index 81447790a..29c127ea8 100644
--- a/cmd/unbounded-storage/ARCHITECTURE.md
+++ b/cmd/unbounded-storage/ARCHITECTURE.md
@@ -42,8 +42,9 @@ serves/relays stripes to and from peer nodes over a libfabric RDMA fabric.
   `!Send`/`!Sync` (`Rc`, `RefCell`, `Cell`, raw pointers) and never cross a
   thread boundary. Cross-core communication is explicit and channel-based.
 - **Topology-driven.** At startup the daemon discovers host hardware (CPUs,
-  NUMA nodes, HCAs, NVMe drives) from sysfs and computes a `Plan` that assigns
-  disjoint CPUs to roles. Everything downstream is sized from that plan.
+  NUMA nodes, HCAs, and safe unused NVMe namespaces) from sysfs and procfs and
+  computes a `Plan` that assigns disjoint CPUs to roles. Everything downstream
+  is sized from that plan.
 - **NUMA locality everywhere.** Memory backings, fabric memory regions, and disk
   engines are all allocated on, and pinned to, the NUMA node of the CPU that
   uses them.
@@ -128,15 +129,19 @@ excluded from the live-reload diff.
 
 ### Startup sequence
 
-1. Load and validate config into one immutable `LoadedConfig` containing the raw
-   config, owned runtime graph, and route snapshot; build `StartupSettings` from
-   its raw config's `[startup]` section.
-2. `Host::discover()` reads hardware from sysfs.
-3. `CorePlan::for_host(&host, &settings.core_plan_config)` partitions the
-   host's usable CPUs into three disjoint, NUMA-local classes: one
-   `StorageCore` per NVMe drive, `nic_workers` `NicWorker`s per active HCA,
-   and a `ServingShard` on every remaining CPU (optionally capped by
-   `serving_cores`).
+1. Load and validate config, then resolve its disk set. A nonempty explicit set
+   is authoritative. Otherwise `disk_discovery` scans for safe unused NVMe
+   namespaces and materializes block `DiskSpec`s; if none remain, it materializes
+   the configured file fallback (20 GiB by default). The resulting immutable
+   `LoadedConfig` contains the resolved raw config, runtime graph, and route
+   snapshot. `StartupSettings` comes from its `[startup]` section.
+2. `Host::discover()` reads CPU, HCA, controller, and inventory hardware from
+   sysfs.
+3. `CorePlan::for_host_with_storage` partitions the host's usable CPUs into
+   three disjoint, NUMA-local classes: one `StorageCore` per initial concrete
+   block disk, `nic_workers` `NicWorker`s per active HCA, and a `ServingShard`
+   on every remaining CPU (optionally capped by `serving_cores`). File disks and
+   block disks added after startup run unpinned.
 4. **One shard thread is spawned per `ServingShard`.** If the host yields no
    serving shards, the daemon exits with failure. With no usable HCA the
    NIC-worker class is simply empty and the shards serve over the
@@ -297,15 +302,16 @@ it a simple loop to drive work without an async runtime.
 
 ### 7.2 `topology/` - hardware discovery and planning
 
-- `Host::discover()` reads CPUs, NUMA nodes, HCAs (`Hca`), NICs (`Nic`), and
-  NVMe drives (`Nvme`) from sysfs.
-- `CorePlan::for_host(&host, &CorePlanConfig)` partitions the host's usable
-  CPUs into three disjoint, NUMA-local classes, scheduled most-constrained
-  first: a `StorageCore` per NVMe drive, then `nic_workers` `NicWorker`s per
-  active HCA (grouped into a `NicWorkerGroup` per HCA), then a `ServingShard`
-  on every remaining CPU. Each CPU is handed out at most once; an exhausted
-  pool oversubscribes rather than panicking. The shared CPU/HCA filtering
-  engine (SMT collapse, isolcpus, cpu0 exclusion, active-port gating) lives in
+- `Host::discover()` reads CPUs, NUMA nodes, HCAs (`Hca`), NICs (`Nic`), NVMe
+  controllers (`Nvme`), and block-device inventory from sysfs. Safe namespace
+  selection is owned separately by `disk_discovery`.
+- `CorePlan::for_host_with_storage` partitions the host's usable CPUs into three
+  disjoint, NUMA-local classes, scheduled most-constrained first: a
+  `StorageCore` per initial block disk, then `nic_workers` `NicWorker`s per
+  active HCA (grouped into a `NicWorkerGroup` per HCA), then a `ServingShard` on
+  every remaining CPU. Each CPU is handed out at most once; an exhausted pool
+  oversubscribes rather than panicking. The shared CPU/HCA filtering engine (SMT
+  collapse, isolcpus, cpu0 exclusion, active-port gating) lives in
   `topology/filters.rs`.
 - `CorePlanConfig` knobs (defaults): `nic_workers` (4 per active HCA),
   `serving_cores` (`None` = claim every remaining CPU), `use_smt_siblings`
@@ -316,7 +322,7 @@ it a simple loop to drive work without an async runtime.
   verbs).
 - Key fields consumed by main: `plan.serving_shards` (one shard thread each),
   `plan.nic_workers` (the fabric worker groups), and `plan.storage_cores`,
-  which main maps to one `DiskCpuSlot` per NVMe drive.
+  which main maps to one `DiskCpuSlot` per initial block disk.
 ### 7.3 `memory/` - NUMA-local backings
 
 - `Backing` is a pinned, NUMA-local memory region carved into fixed-size pages
@@ -660,19 +666,32 @@ Sections (all optional, each falling back to defaults):
   `config` table (`block` with `path` and optional `numa`, or `file` with
   `path` and required `size`), `queue_depth` (optional), `page_size_bytes`, and
   `skip_recovery_scan` (fields that disk reconcile treats as drift, see 7.10).
-  Disk paths must be unique across the shared set.
+  Disk paths must be unique across the shared set. A nonempty explicit set is
+  authoritative and bypasses automatic discovery and its deny list.
+- `[disk_discovery]` - policy used only when `[[disks]]` is empty.
+  `denied_paths` contains exact absolute device paths and filters automatic
+  selection only. A whole NVMe namespace is eligible when it has nonzero
+  capacity, its `/dev` identity matches sysfs, and neither it nor any child
+  partition is mounted, active swap, or has holders. Mount, swap, identity, and
+  holder checks use device major:minor and fail closed; an unmounted partition
+  table alone is allowed. Optional `[disk_discovery.fallback]` reuses the file
+  disk shape and defaults missing fields to
+  `/var/lib/unbounded-storage/cache.disk` and 20 GiB.
 - `[[frontends]]` - `name`, `source` (a backend or cache component name), and
   one `config` table (`http`, `s3`, or `loadgen`).
 
-After defaults and validation, loading constructs one immutable `LoadedConfig`:
-the raw `Config`, one owned `RuntimeGraph`, and one route snapshot built from
-that graph. The watcher (`notify`-based) emits these loaded snapshots; main's
-`wait_for_shutdown_with_updates` reconciles peers (remove + add on address/numa
-drift, via a `last_applied` cache), disks, and - by broadcasting the applied
-config to every shard - each shard's backend and frontend registries plus the
-routing snapshot. The apply target and shards consume the same loaded graph and
-routes; this is coherent preparation, not whole-process transactionality. It
-republishes the channel snapshot each update, logs
+After defaults and validation, main resolves disks and rebuilds one immutable
+`LoadedConfig`: the resolved raw `Config`, one owned `RuntimeGraph`, and one
+route snapshot built from that graph. The resolver reuses an automatic result
+while the discovery policy is unchanged. It rescans on a policy change or an
+explicit-to-automatic transition; scan failure keeps the prior live config.
+The watcher (`notify`-based) emits unresolved loaded snapshots; main resolves
+them before `wait_for_shutdown_with_updates` reconciles peers (remove + add on
+address/numa drift, via a `last_applied` cache), disks, and - by broadcasting the
+applied config to every shard - each shard's backend and frontend registries
+plus the routing snapshot. The apply target and shards consume the same loaded
+graph and routes; this is coherent preparation, not whole-process
+transactionality. It republishes the channel snapshot each update, logs
 `config gen=N ...`, and sets `SHUTDOWN` if the watcher disconnects.
 
 ### 7.12 `tls/` - the shared TLS transport
diff --git a/cmd/unbounded-storage/README.md b/cmd/unbounded-storage/README.md
index edcc9f51e..13a6365c1 100644
--- a/cmd/unbounded-storage/README.md
+++ b/cmd/unbounded-storage/README.md
@@ -36,8 +36,8 @@ The built binary is copied to `bin/unbounded-storage`.
 
 ```bash
 # default: read /etc/unbounded-storage/config.toml; if absent, fall
-# back to built-in defaults (a heap-backed, single-shard, no-peer,
-# no-disk run that is mostly useful as a smoke test).
+# back to built-in defaults. Safe unused NVMe namespaces are selected
+# automatically; if none are available, a 20 GiB file disk is used.
 unbounded-storage
 
 # explicit config path; missing or invalid here is fatal
@@ -142,6 +142,8 @@ addr      = "10.0.0.1:9000"      # parsed as SocketAddr.
 name = "cache"
 source = "origin"               # backend used for miss fills.
 
+# Optional explicit disk list. When nonempty it is authoritative and automatic
+# discovery, including denied_paths, does not apply.
 [[disks]]                        # repeat per local device; paths must be unique.
 queue_depth = 32                 # optional u32; per-disk io_uring depth.
 skip_recovery_scan = false       # only for fresh or benchmark disks.
@@ -155,6 +157,19 @@ numa        = 0                  # optional u16; biases the open onto a CPU on t
 # path = "/var/lib/unbounded-storage/disk0.img"
 # size = 1073741824              # required bytes for file-backed disks.
 
+# When [[disks]] is absent or empty, whole NVMe namespaces are selected if the
+# namespace and all child partitions are not mounted, active swap, or held by
+# LVM/RAID/device-mapper. Safety state is checked by device major:minor and
+# discovery fails closed when it cannot be established. An unmounted partition
+# table alone does not exclude a namespace.
+[disk_discovery]
+denied_paths = ["/dev/nvme1n1"] # exact paths; automatic selection only.
+
+# Used only when no eligible automatic disks remain. Both fields are optional.
+[disk_discovery.fallback]
+path = "/var/lib/unbounded-storage/cache.disk"
+size = 21474836480               # 20 GiB default.
+
 [[frontends]]
 name = "http"
 source = "cache"                # backend or cache component name.
@@ -193,6 +208,12 @@ allow_inactive_port   = false    # use HCA ports not in the active state.
 disable_rdma          = false    # disable RDMA and force the libfabric tcp provider.
 ```
 
+Automatic selections are cached while `disk_discovery` is unchanged. Changing
+that policy, or changing from explicit disks back to automatic mode, rescans the
+host before applying the reload. Explicit disks remain dynamically reloadable.
+CPU slots are planned for the initial block-disk set; block disks introduced by
+a later reload run unpinned until the daemon restarts.
+
 ### Origin authentication and TLS
 
 HTTP, S3, and Azure backend URLs must contain only a scheme and authority, with
diff --git a/cmd/unbounded-storage/build.rs b/cmd/unbounded-storage/build.rs
index f16ddad8e..5903d4309 100644
--- a/cmd/unbounded-storage/build.rs
+++ b/cmd/unbounded-storage/build.rs
@@ -103,6 +103,7 @@ fn generate_config_schema() {
         "DiskSpec",
         "BlockDiskConfig",
         "FileDiskConfig",
+        "DiskDiscoveryCfg",
         "CacheSpec",
         "BackendSpec",
         "HttpBackendConfig",
diff --git a/cmd/unbounded-storage/src/config/load.rs b/cmd/unbounded-storage/src/config/load.rs
index f2d290767..5679fe369 100644
--- a/cmd/unbounded-storage/src/config/load.rs
+++ b/cmd/unbounded-storage/src/config/load.rs
@@ -93,6 +93,8 @@ pub enum ConfigError {
     },
     Protobuf(prost::DecodeError),
     DuplicateDiskPath(String),
+    DuplicateDeniedDiskPath(String),
+    InvalidDeniedDiskPath(String),
     MissingFileDiskSize(String),
     ZeroFileDiskSize(String),
     FileDiskSizeNotPageMultiple {
@@ -190,6 +192,15 @@ impl fmt::Display for ConfigError {
             ConfigError::DuplicateDiskPath(p) => {
                 write!(f, "duplicate disk path: {p}")
             }
+            ConfigError::DuplicateDeniedDiskPath(p) => {
+                write!(f, "duplicate denied disk path: {p}")
+            }
+            ConfigError::InvalidDeniedDiskPath(p) => {
+                write!(
+                    f,
+                    "denied disk path must be an absolute path below /dev: {p}"
+                )
+            }
             ConfigError::MissingFileDiskSize(p) => {
                 write!(f, "disk {p}: file config requires a `size`")
             }
@@ -474,6 +485,7 @@ fn validate(cfg: &Config) -> Result<(), ConfigError> {
             return Err(ConfigError::DuplicateDiskPath(path.to_string()));
         }
     }
+    validate_disk_discovery(cfg.disk_discovery())?;
 
     validate_mesh(cfg)?;
 
@@ -747,20 +759,7 @@ fn validate_disks(disks: &[super::schema::DiskSpec]) -> Result<(), ConfigError>
                     return Err(ConfigError::DuplicateDiskPath(path.to_string()));
                 }
                 let page_size = d.page_size_bytes.unwrap_or(4096);
-                let size = match cfg.size {
-                    Some(s) => s,
-                    None => return Err(ConfigError::MissingFileDiskSize(path.to_string())),
-                };
-                if size == 0 {
-                    return Err(ConfigError::ZeroFileDiskSize(path.to_string()));
-                }
-                if page_size == 0 || size % page_size != 0 {
-                    return Err(ConfigError::FileDiskSizeNotPageMultiple {
-                        path: path.to_string(),
-                        size,
-                        page_size,
-                    });
-                }
+                validate_file_disk(path, cfg.size, page_size)?;
             }
             Some(disk_spec::Config::Block(_)) => {
                 let path = validated_disk_path(d)?;
@@ -775,6 +774,52 @@ fn validate_disks(disks: &[super::schema::DiskSpec]) -> Result<(), ConfigError>
     Ok(())
 }
 
+fn validate_disk_discovery(discovery: &super::schema::DiskDiscoveryCfg) -> Result<(), ConfigError> {
+    let mut seen_paths = HashSet::new();
+    for path in &discovery.denied_paths {
+        let parsed = Path::new(path);
+        if !parsed.is_absolute()
+            || !parsed.starts_with("/dev")
+            || parsed == Path::new("/dev")
+            || parsed
+                .components()
+                .any(|component| matches!(component, std::path::Component::ParentDir))
+        {
+            return Err(ConfigError::InvalidDeniedDiskPath(path.clone()));
+        }
+        if !seen_paths.insert(path.as_str()) {
+            return Err(ConfigError::DuplicateDeniedDiskPath(path.clone()));
+        }
+    }
+
+    let fallback = discovery
+        .fallback
+        .as_ref()
+        .expect("disk discovery fallback populated");
+    validate_file_disk(&fallback.path, fallback.size, 4096)
+}
+
+fn validate_file_disk(path: &str, size: Option<u64>, page_size: u64) -> Result<(), ConfigError> {
+    if path.is_empty() {
+        return Err(ConfigError::EmptyDiskPath);
+    }
+    let Some(size) = size else {
+        return Err(ConfigError::MissingFileDiskSize(path.to_string()));
+    };
+    if size == 0 {
+        return Err(ConfigError::ZeroFileDiskSize(path.to_string()));
+    }
+    if page_size == 0 || size % page_size != 0 {
+        return Err(ConfigError::FileDiskSizeNotPageMultiple {
+            path: path.to_string(),
+            size,
+            page_size,
+        });
+    }
+
+    Ok(())
+}
+
 fn validated_disk_path(disk: &super::schema::DiskSpec) -> Result<&str, ConfigError> {
     let Some(path) = disk.path() else {
         return Err(ConfigError::MissingDiskConfig);
@@ -886,6 +931,95 @@ url = "https://s3.example.com"
         let cfg = cfg.config();
         assert!(cfg.peers.is_empty());
         assert!(cfg.caches.is_empty());
+        assert_eq!(
+            cfg.disk_discovery().fallback.as_ref().unwrap().path,
+            "/var/lib/unbounded-storage/cache.disk"
+        );
+        assert_eq!(
+            cfg.disk_discovery().fallback.as_ref().unwrap().size,
+            Some(20 * 1024 * 1024 * 1024)
+        );
+    }
+
+    #[test]
+    fn defaults_partial_disk_discovery_fallback() {
+        let f = write_cfg(
+            r#"
+[disk_discovery]
+denied_paths = ["/dev/nvme1n1"]
+
+[disk_discovery.fallback]
+path = "/var/cache/unbounded-storage.disk"
+"#,
+        );
+        let cfg = load(f.path()).expect("load should succeed");
+        let discovery = cfg.config().disk_discovery();
+        assert_eq!(discovery.denied_paths, ["/dev/nvme1n1"]);
+        assert_eq!(
+            discovery.fallback.as_ref().unwrap().path,
+            "/var/cache/unbounded-storage.disk"
+        );
+        assert_eq!(
+            discovery.fallback.as_ref().unwrap().size,
+            Some(20 * 1024 * 1024 * 1024)
+        );
+    }
+
+    #[test]
+    fn defaults_partial_disk_discovery_fallback_path() {
+        let f = write_cfg(
+            r#"
+[disk_discovery.fallback]
+size = 1073741824
+"#,
+        );
+        let cfg = load(f.path()).expect("load should succeed");
+        let fallback = cfg.config().disk_discovery().fallback.as_ref().unwrap();
+        assert_eq!(fallback.path, "/var/lib/unbounded-storage/cache.disk");
+        assert_eq!(fallback.size, Some(1024 * 1024 * 1024));
+    }
+
+    #[test]
+    fn rejects_relative_denied_disk_path() {
+        let f = write_cfg(
+            r#"
+[disk_discovery]
+denied_paths = ["nvme0n1"]
+"#,
+        );
+        assert!(matches!(
+            load(f.path()),
+            Err(ConfigError::InvalidDeniedDiskPath(path)) if path == "nvme0n1"
+        ));
+    }
+
+    #[test]
+    fn rejects_duplicate_denied_disk_path() {
+        let f = write_cfg(
+            r#"
+[disk_discovery]
+denied_paths = ["/dev/nvme0n1", "/dev/nvme0n1"]
+"#,
+        );
+        assert!(matches!(
+            load(f.path()),
+            Err(ConfigError::DuplicateDeniedDiskPath(path)) if path == "/dev/nvme0n1"
+        ));
+    }
+
+    #[test]
+    fn validates_disk_discovery_fallback() {
+        let f = write_cfg(
+            r#"
+[disk_discovery.fallback]
+size = 0
+"#,
+        );
+        assert!(matches!(
+            load(f.path()),
+            Err(ConfigError::ZeroFileDiskSize(path))
+                if path == "/var/lib/unbounded-storage/cache.disk"
+        ));
     }
 
     #[test]
diff --git a/cmd/unbounded-storage/src/config/mod.rs b/cmd/unbounded-storage/src/config/mod.rs
index 5e74be7c5..6f17ebfd0 100644
--- a/cmd/unbounded-storage/src/config/mod.rs
+++ b/cmd/unbounded-storage/src/config/mod.rs
@@ -31,10 +31,10 @@ pub use reconcile::{
 };
 pub use schema::{
     AutoRdmaFabricBinds, AzureBackendConfig, BackendSpec, BlockDiskConfig, CacheSpec, Config,
-    DiskSpec, FabricCfg, FakeBackendConfig, FileDiskConfig, FrontendSpec, HttpBackendConfig,
-    HttpFrontendConfig, LoadgenFrontendConfig, MemoryCfg, PeerSpec, RdmaPeerConfig, RoutingPlan,
-    S3BackendConfig, S3FrontendConfig, StartupCfg, TcpFabricBinds, TcpPeerConfig, TopologyCfg,
-    TopologyPrefixWeight, TopologyWeighting, backend_spec, disk_spec, fabric_cfg, frontend_spec,
-    peer_spec,
+    DiskDiscoveryCfg, DiskSpec, FabricCfg, FakeBackendConfig, FileDiskConfig, FrontendSpec,
+    HttpBackendConfig, HttpFrontendConfig, LoadgenFrontendConfig, MemoryCfg, PeerSpec,
+    RdmaPeerConfig, RoutingPlan, S3BackendConfig, S3FrontendConfig, StartupCfg, TcpFabricBinds,
+    TcpPeerConfig, TopologyCfg, TopologyPrefixWeight, TopologyWeighting, backend_spec, disk_spec,
+    fabric_cfg, frontend_spec, peer_spec,
 };
 pub use watch::{ConfigUpdate, ConfigWatcher, WatchError};
diff --git a/cmd/unbounded-storage/src/config/schema.rs b/cmd/unbounded-storage/src/config/schema.rs
index fdc475016..55c70c526 100644
--- a/cmd/unbounded-storage/src/config/schema.rs
+++ b/cmd/unbounded-storage/src/config/schema.rs
@@ -25,6 +25,8 @@ const DEFAULT_STRIPE_SIZE_BYTES: u64 = 4 * 1024 * 1024;
 const DEFAULT_HTTP_CONCURRENCY: u32 = 64;
 pub const DEFAULT_HTTP_FRONTEND_MAX_REQUESTS_PER_CONNECTION: u32 = 1024;
 const DEFAULT_FAKE_OBJECT_SIZE_BYTES: u64 = 1024 * 1024;
+pub const DEFAULT_DISK_FALLBACK_PATH: &str = "/var/lib/unbounded-storage/cache.disk";
+pub const DEFAULT_DISK_FALLBACK_SIZE_BYTES: u64 = 20 * 1024 * 1024 * 1024;
 
 impl Config {
     /// Populates every omitted section and optional defaulted field with
@@ -42,6 +44,19 @@ impl Config {
             frontend.apply_defaults();
         }
 
+        let disk_discovery = self
+            .disk_discovery
+            .get_or_insert_with(DiskDiscoveryCfg::default);
+        let fallback = disk_discovery
+            .fallback
+            .get_or_insert_with(FileDiskConfig::default);
+        if fallback.path.is_empty() {
+            fallback.path = DEFAULT_DISK_FALLBACK_PATH.to_string();
+        }
+        fallback
+            .size
+            .get_or_insert(DEFAULT_DISK_FALLBACK_SIZE_BYTES);
+
         let startup = self.startup.get_or_insert_with(StartupCfg::default);
 
         let memory = startup.memory.get_or_insert_with(MemoryCfg::default);
@@ -84,6 +99,13 @@ impl Config {
     pub fn startup(&self) -> &StartupCfg {
         self.startup.as_ref().expect("startup section populated")
     }
+
+    /// Automatic disk discovery policy. Valid after [`Config::apply_defaults`].
+    pub fn disk_discovery(&self) -> &DiskDiscoveryCfg {
+        self.disk_discovery
+            .as_ref()
+            .expect("disk discovery section populated")
+    }
 }
 
 impl BackendSpec {
diff --git a/cmd/unbounded-storage/src/disk_discovery.rs b/cmd/unbounded-storage/src/disk_discovery.rs
new file mode 100644
index 000000000..dffa6973c
--- /dev/null
+++ b/cmd/unbounded-storage/src/disk_discovery.rs
@@ -0,0 +1,829 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+use std::collections::{BTreeSet, HashSet};
+use std::fmt;
+use std::fs;
+use std::io;
+use std::os::unix::fs::{FileTypeExt, MetadataExt};
+use std::path::{Path, PathBuf};
+
+use crate::config::{self, BlockDiskConfig, Config, DiskDiscoveryCfg, DiskSpec};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct DeviceId {
+    pub major: u32,
+    pub minor: u32,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AutoDisk {
+    pub path: String,
+    pub numa: Option<u16>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ExclusionReason {
+    Denied,
+    Mounted,
+    Swap,
+    Held,
+    ZeroCapacity,
+    DeviceIdentityMismatch,
+    SafetyStateUnavailable,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ExcludedDisk {
+    pub path: String,
+    pub reason: ExclusionReason,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DiscoveryReport {
+    pub eligible: Vec<AutoDisk>,
+    pub excluded: Vec<ExcludedDisk>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum DiskSource {
+    Explicit,
+    Automatic,
+    Fallback,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct DiskResolution {
+    pub source: DiskSource,
+    pub report: Option<DiscoveryReport>,
+}
+
+#[derive(Default)]
+pub struct DiskResolver {
+    cached: Option<CachedResolution>,
+}
+
+impl DiskResolver {
+    pub fn resolve(&mut self, config: &mut Config) -> Result<DiskResolution, DiscoveryError> {
+        self.resolve_with(config, discover)
+    }
+
+    pub fn resolve_with(
+        &mut self,
+        config: &mut Config,
+        mut scan: impl FnMut(&[String]) -> Result<DiscoveryReport, DiscoveryError>,
+    ) -> Result<DiskResolution, DiscoveryError> {
+        if !config.disks.is_empty() {
+            self.cached = None;
+            return Ok(DiskResolution {
+                source: DiskSource::Explicit,
+                report: None,
+            });
+        }
+
+        let policy = config.disk_discovery().clone();
+        if let Some(cached) = self
+            .cached
+            .as_ref()
+            .filter(|cached| cached.policy == policy)
+        {
+            config.disks.clone_from(&cached.disks);
+            return Ok(cached.resolution.clone());
+        }
+
+        let report = scan(&policy.denied_paths)?;
+        let (source, disks) = if report.eligible.is_empty() {
+            (DiskSource::Fallback, vec![fallback_spec(&policy)])
+        } else {
+            (
+                DiskSource::Automatic,
+                report.eligible.iter().map(block_spec).collect(),
+            )
+        };
+        let resolution = DiskResolution {
+            source,
+            report: Some(report),
+        };
+        config.disks.clone_from(&disks);
+        self.cached = Some(CachedResolution {
+            policy,
+            disks,
+            resolution: resolution.clone(),
+        });
+        Ok(resolution)
+    }
+}
+
+struct CachedResolution {
+    policy: DiskDiscoveryCfg,
+    disks: Vec<DiskSpec>,
+    resolution: DiskResolution,
+}
+
+#[derive(Clone, Debug)]
+pub struct DiscoveryRoots {
+    pub sys: PathBuf,
+    pub proc: PathBuf,
+    pub dev: PathBuf,
+}
+
+impl DiscoveryRoots {
+    pub fn host() -> Self {
+        Self {
+            sys: PathBuf::from("/sys"),
+            proc: PathBuf::from("/proc"),
+            dev: PathBuf::from("/dev"),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum DiscoveryError {
+    ReadBlockDevices(io::Error),
+    ReadMountInfo(io::Error),
+    ReadSwaps(io::Error),
+    InvalidMountInfo(String),
+    InvalidSwapDevice(String),
+}
+
+impl fmt::Display for DiscoveryError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::ReadBlockDevices(error) => write!(f, "read block devices: {error}"),
+            Self::ReadMountInfo(error) => write!(f, "read mount information: {error}"),
+            Self::ReadSwaps(error) => write!(f, "read swap information: {error}"),
+            Self::InvalidMountInfo(line) => write!(f, "invalid mountinfo entry: {line}"),
+            Self::InvalidSwapDevice(path) => write!(f, "cannot identify swap device: {path}"),
+        }
+    }
+}
+
+impl std::error::Error for DiscoveryError {}
+
+pub trait DeviceProbe {
+    fn identity(&self, path: &Path) -> io::Result<Option<DeviceId>>;
+
+    fn backing_identity(&self, path: &Path) -> io::Result<DeviceId>;
+}
+
+pub struct SystemDeviceProbe;
+
+impl DeviceProbe for SystemDeviceProbe {
+    fn identity(&self, path: &Path) -> io::Result<Option<DeviceId>> {
+        let metadata = fs::metadata(path)?;
+        if !metadata.file_type().is_block_device() {
+            return Ok(None);
+        }
+        let device = metadata.rdev();
+        Ok(Some(DeviceId {
+            major: libc::major(device) as u32,
+            minor: libc::minor(device) as u32,
+        }))
+    }
+
+    fn backing_identity(&self, path: &Path) -> io::Result<DeviceId> {
+        let device = fs::metadata(path)?.dev();
+        Ok(DeviceId {
+            major: libc::major(device) as u32,
+            minor: libc::minor(device) as u32,
+        })
+    }
+}
+
+#[derive(Debug)]
+struct BlockEntry {
+    name: String,
+    id: Option<DeviceId>,
+    partition: bool,
+}
+
+pub fn discover(denied_paths: &[String]) -> Result<DiscoveryReport, DiscoveryError> {
+    discover_with(DiscoveryRoots::host(), &SystemDeviceProbe, denied_paths)
+}
+
+pub fn discover_with(
+    roots: DiscoveryRoots,
+    device_probe: &impl DeviceProbe,
+    denied_paths: &[String],
+) -> Result<DiscoveryReport, DiscoveryError> {
+    let mount_ids = read_mount_ids(&roots.proc.join("self/mountinfo"))?;
+    let swap_ids = read_swap_ids(&roots, device_probe)?;
+    let block_entries = read_block_entries(&roots.sys.join("class/block"))?;
+    let denied = denied_paths
+        .iter()
+        .map(String::as_str)
+        .collect::<HashSet<_>>();
+    let mut report = DiscoveryReport::default();
+
+    for entry in block_entries
+        .iter()
+        .filter(|entry| !entry.partition && nvme_controller(&entry.name).is_some())
+    {
+        let path = format!("/dev/{}", entry.name);
+        let descendants = block_entries
+            .iter()
+            .filter(|child| child.partition && is_partition_of(&child.name, &entry.name))
+            .collect::<Vec<_>>();
+        let mut identities = descendants
+            .iter()
+            .filter_map(|child| child.id)
+            .collect::<BTreeSet<_>>();
+        if let Some(id) = entry.id {
+            identities.insert(id);
+        }
+        let sectors = read_sectors(&roots.sys.join("class/block").join(&entry.name).join("size"));
+
+        let reason = if denied.contains(path.as_str()) {
+            Some(ExclusionReason::Denied)
+        } else if entry.id.is_none() || descendants.iter().any(|child| child.id.is_none()) {
+            Some(ExclusionReason::SafetyStateUnavailable)
+        } else if sectors == Some(0) {
+            Some(ExclusionReason::ZeroCapacity)
+        } else if device_probe
+            .identity(&roots.dev.join(&entry.name))
+            .ok()
+            .flatten()
+            != entry.id
+        {
+            Some(ExclusionReason::DeviceIdentityMismatch)
+        } else if identities.iter().any(|id| mount_ids.contains(id)) {
+            Some(ExclusionReason::Mounted)
+        } else if identities.iter().any(|id| swap_ids.contains(id)) {
+            Some(ExclusionReason::Swap)
+        } else if has_holders(&roots.sys, &entry.name)
+            || descendants
+                .iter()
+                .any(|child| has_holders(&roots.sys, &child.name))
+        {
+            Some(ExclusionReason::Held)
+        } else if sectors.is_none() {
+            Some(ExclusionReason::SafetyStateUnavailable)
+        } else {
+            None
+        };
+
+        if let Some(reason) = reason {
+            report.excluded.push(ExcludedDisk { path, reason });
+            continue;
+        }
+
+        let controller = nvme_controller(&entry.name).expect("validated namespace name");
+        report.eligible.push(AutoDisk {
+            path,
+            numa: read_numa(
+                &roots
+                    .sys
+                    .join("class/nvme")
+                    .join(controller)
+                    .join("device/numa_node"),
+            ),
+        });
+    }
+
+    report.eligible.sort_by(|a, b| a.path.cmp(&b.path));
+    report.excluded.sort_by(|a, b| a.path.cmp(&b.path));
+    Ok(report)
+}
+
+fn read_block_entries(path: &Path) -> Result<Vec<BlockEntry>, DiscoveryError> {
+    let entries = fs::read_dir(path).map_err(DiscoveryError::ReadBlockDevices)?;
+    let mut blocks = Vec::new();
+    for entry in entries.flatten() {
+        let Some(name) = entry.file_name().to_str().map(str::to_owned) else {
+            continue;
+        };
+        blocks.push(BlockEntry {
+            id: fs::read_to_string(entry.path().join("dev"))
+                .ok()
+                .and_then(|value| parse_device_id(value.trim())),
+            partition: entry.path().join("partition").exists(),
+            name,
+        });
+    }
+    blocks.sort_by(|a, b| a.name.cmp(&b.name));
+    Ok(blocks)
+}
+
+fn read_mount_ids(path: &Path) -> Result<HashSet<DeviceId>, DiscoveryError> {
+    let contents = fs::read_to_string(path).map_err(DiscoveryError::ReadMountInfo)?;
+    contents
+        .lines()
+        .map(|line| {
+            line.split_whitespace()
+                .nth(2)
+                .and_then(parse_device_id)
+                .ok_or_else(|| DiscoveryError::InvalidMountInfo(line.to_string()))
+        })
+        .collect()
+}
+
+fn read_swap_ids(
+    roots: &DiscoveryRoots,
+    device_probe: &impl DeviceProbe,
+) -> Result<HashSet<DeviceId>, DiscoveryError> {
+    let contents =
+        fs::read_to_string(roots.proc.join("swaps")).map_err(DiscoveryError::ReadSwaps)?;
+    let mut identities = HashSet::new();
+    for line in contents.lines().skip(1) {
+        let Some(path) = line.split_whitespace().next() else {
+            continue;
+        };
+        let swap_path = Path::new(path);
+        let rooted = swap_path
+            .strip_prefix("/dev")
+            .map(|relative| roots.dev.join(relative))
+            .unwrap_or_else(|_| roots.proc.join("root").join(path.trim_start_matches('/')));
+        let id = match device_probe.identity(&rooted) {
+            Ok(Some(id)) => Ok(id),
+            Ok(None) => device_probe.backing_identity(&rooted),
+            Err(error) => Err(error),
+        }
+        .map_err(|_| DiscoveryError::InvalidSwapDevice(path.to_string()))?;
+        identities.insert(id);
+    }
+    Ok(identities)
+}
+
+fn has_holders(sys_root: &Path, name: &str) -> bool {
+    fs::read_dir(sys_root.join("class/block").join(name).join("holders"))
+        .map(|mut entries| entries.next().is_some())
+        .unwrap_or(true)
+}
+
+fn read_sectors(path: &Path) -> Option<u64> {
+    fs::read_to_string(path).ok()?.trim().parse().ok()
+}
+
+fn read_numa(path: &Path) -> Option<u16> {
+    let value = fs::read_to_string(path).ok()?.trim().parse::<i32>().ok()?;
+    u16::try_from(value).ok()
+}
+
+fn parse_device_id(value: &str) -> Option<DeviceId> {
+    let (major, minor) = value.split_once(':')?;
+    Some(DeviceId {
+        major: major.parse().ok()?,
+        minor: minor.parse().ok()?,
+    })
+}
+
+fn nvme_controller(name: &str) -> Option<&str> {
+    let suffix = name.strip_prefix("nvme")?;
+    let controller_digits = suffix.chars().take_while(char::is_ascii_digit).count();
+    if controller_digits == 0 {
+        return None;
+    }
+    let namespace = suffix.get(controller_digits..)?.strip_prefix('n')?;
+    if namespace.is_empty() || !namespace.chars().all(|c| c.is_ascii_digit()) {
+        return None;
+    }
+    name.get(.."nvme".len() + controller_digits)
+}
+
+fn is_partition_of(name: &str, namespace: &str) -> bool {
+    name.strip_prefix(namespace)
+        .and_then(|suffix| suffix.strip_prefix('p'))
+        .is_some_and(|number| !number.is_empty() && number.chars().all(|c| c.is_ascii_digit()))
+}
+
+fn block_spec(disk: &AutoDisk) -> DiskSpec {
+    DiskSpec {
+        config: Some(config::disk_spec::Config::Block(BlockDiskConfig {
+            path: disk.path.clone(),
+            numa: disk.numa.map(u32::from),
+        })),
+        ..DiskSpec::default()
+    }
+}
+
+fn fallback_spec(policy: &DiskDiscoveryCfg) -> DiskSpec {
+    DiskSpec {
+        config: Some(config::disk_spec::Config::File(
+            policy
+                .fallback
+                .clone()
+                .expect("fallback populated by Config::apply_defaults"),
+        )),
+        ..DiskSpec::default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::cell::Cell;
+    use std::collections::HashMap;
+    use std::fs;
+    use std::io;
+    use std::path::{Path, PathBuf};
+
+    use super::*;
+
+    struct Fixture {
+        root: tempfile::TempDir,
+        devices: HashMap<PathBuf, DeviceId>,
+    }
+
+    impl Fixture {
+        fn new() -> Self {
+            let fixture = Self {
+                root: tempfile::tempdir().expect("create fixture"),
+                devices: HashMap::new(),
+            };
+            fixture.write("proc/self/mountinfo", "");
+            fixture.write("proc/swaps", "Filename\tType\tSize\tUsed\tPriority\n");
+            fixture
+        }
+
+        fn roots(&self) -> DiscoveryRoots {
+            DiscoveryRoots {
+                sys: self.root.path().join("sys"),
+                proc: self.root.path().join("proc"),
+                dev: self.root.path().join("dev"),
+            }
+        }
+
+        fn write(&self, relative: &str, contents: &str) {
+            let path = self.root.path().join(relative);
+            fs::create_dir_all(path.parent().expect("fixture file parent"))
+                .expect("create fixture directory");
+            fs::write(path, contents).expect("write fixture file");
+        }
+
+        fn add_namespace(&mut self, name: &str, id: DeviceId, sectors: u64, numa: i16) {
+            self.write(
+                &format!("sys/class/block/{name}/dev"),
+                &format!("{}:{}\n", id.major, id.minor),
+            );
+            self.write(
+                &format!("sys/class/block/{name}/size"),
+                &format!("{sectors}\n"),
+            );
+            fs::create_dir_all(
+                self.root
+                    .path()
+                    .join(format!("sys/class/block/{name}/holders")),
+            )
+            .expect("create holders directory");
+            let controller = nvme_controller(name).expect("NVMe namespace name");
+            self.write(
+                &format!("sys/class/nvme/{controller}/device/numa_node"),
+                &format!("{numa}\n"),
+            );
+            let device_path = self.root.path().join("dev").join(name);
+            self.write(&format!("dev/{name}"), "");
+            self.devices.insert(device_path, id);
+        }
+
+        fn add_partition(&mut self, namespace: &str, number: u16, id: DeviceId) -> String {
+            let name = format!("{namespace}p{number}");
+            self.write(
+                &format!("sys/class/block/{name}/partition"),
+                &format!("{number}\n"),
+            );
+            self.write(
+                &format!("sys/class/block/{name}/dev"),
+                &format!("{}:{}\n", id.major, id.minor),
+            );
+            fs::create_dir_all(
+                self.root
+                    .path()
+                    .join(format!("sys/class/block/{name}/holders")),
+            )
+            .expect("create partition holders directory");
+            let device_path = self.root.path().join("dev").join(&name);
+            self.write(&format!("dev/{name}"), "");
+            self.devices.insert(device_path, id);
+            name
+        }
+
+        fn add_holder(&self, device: &str, holder: &str) {
+            fs::create_dir_all(
+                self.root
+                    .path()
+                    .join(format!("sys/class/block/{device}/holders/{holder}")),
+            )
+            .expect("create holder");
+        }
+    }
+
+    impl DeviceProbe for Fixture {
+        fn identity(&self, path: &Path) -> io::Result<Option<DeviceId>> {
+            Ok(self.devices.get(path).copied())
+        }
+
+        fn backing_identity(&self, path: &Path) -> io::Result<DeviceId> {
+            self.devices
+                .get(path)
+                .copied()
+                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "unknown backing device"))
+        }
+    }
+
+    fn id(major: u32, minor: u32) -> DeviceId {
+        DeviceId { major, minor }
+    }
+
+    #[test]
+    fn discovers_safe_nvme_namespaces_in_path_order() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme10n2", id(259, 2), 16, -1);
+        fixture.add_namespace("nvme2n1", id(259, 1), 8, 3);
+        fixture.add_partition("nvme2n1", 1, id(259, 3));
+        fixture.write("sys/class/block/sda/dev", "8:0\n");
+
+        let report = discover_with(fixture.roots(), &fixture, &[]).expect("discover disks");
+
+        assert_eq!(
+            report.eligible,
+            vec![
+                AutoDisk {
+                    path: "/dev/nvme10n2".to_string(),
+                    numa: None,
+                },
+                AutoDisk {
+                    path: "/dev/nvme2n1".to_string(),
+                    numa: Some(3),
+                },
+            ]
+        );
+        assert!(report.excluded.is_empty());
+    }
+
+    #[test]
+    fn excludes_mounted_swap_held_denied_and_invalid_namespaces() {
+        let mut fixture = Fixture::new();
+        for index in 0..8 {
+            fixture.add_namespace(
+                &format!("nvme{index}n1"),
+                id(259, index),
+                if index == 7 { 0 } else { 8 },
+                0,
+            );
+        }
+        let mounted_partition = fixture.add_partition("nvme2n1", 1, id(259, 20));
+        let swap_partition = fixture.add_partition("nvme3n1", 1, id(259, 30));
+        fixture.add_partition("nvme6n1", 1, id(259, 60));
+        fixture.add_holder("nvme4n1", "dm-0");
+        fixture
+            .devices
+            .insert(fixture.root.path().join("dev/nvme6n1"), id(259, 99));
+        fixture.write(
+            "proc/self/mountinfo",
+            &format!(
+                "20 1 259:1 / /mnt/direct rw - ext4 /dev/nvme1n1 rw\n21 1 259:20 / /mnt/partition rw - ext4 /dev/{mounted_partition} rw\n"
+            ),
+        );
+        fixture.write(
+            "proc/swaps",
+            &format!(
+                "Filename\tType\tSize\tUsed\tPriority\n/dev/{swap_partition}\tpartition\t1024\t0\t-2\n"
+            ),
+        );
+
+        let report = discover_with(fixture.roots(), &fixture, &["/dev/nvme5n1".to_string()])
+            .expect("discover disks");
+
+        assert_eq!(
+            report.eligible,
+            vec![AutoDisk {
+                path: "/dev/nvme0n1".to_string(),
+                numa: Some(0),
+            }]
+        );
+        assert_eq!(
+            report
+                .excluded
+                .iter()
+                .map(|disk| (disk.path.as_str(), disk.reason))
+                .collect::<Vec<_>>(),
+            vec![
+                ("/dev/nvme1n1", ExclusionReason::Mounted),
+                ("/dev/nvme2n1", ExclusionReason::Mounted),
+                ("/dev/nvme3n1", ExclusionReason::Swap),
+                ("/dev/nvme4n1", ExclusionReason::Held),
+                ("/dev/nvme5n1", ExclusionReason::Denied),
+                ("/dev/nvme6n1", ExclusionReason::DeviceIdentityMismatch),
+                ("/dev/nvme7n1", ExclusionReason::ZeroCapacity),
+            ]
+        );
+    }
+
+    #[test]
+    fn swap_file_excludes_its_backing_namespace() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme0n1", id(259, 0), 8, 0);
+        let swap_file = fixture.root.path().join("proc/root/swapfile");
+        fixture.write("proc/root/swapfile", "");
+        fixture.devices.insert(swap_file, id(259, 0));
+        fixture.write(
+            "proc/swaps",
+            "Filename\tType\tSize\tUsed\tPriority\n/swapfile\tfile\t1024\t0\t-2\n",
+        );
+
+        let report = discover_with(fixture.roots(), &fixture, &[]).expect("discover disks");
+
+        assert!(report.eligible.is_empty());
+        assert_eq!(report.excluded[0].reason, ExclusionReason::Swap);
+    }
+
+    #[test]
+    fn unresolved_swap_file_fails_closed() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme0n1", id(259, 0), 8, 0);
+        fixture.write(
+            "proc/swaps",
+            "Filename\tType\tSize\tUsed\tPriority\n/missing.swap\tfile\t1024\t0\t-2\n",
+        );
+
+        assert!(matches!(
+            discover_with(fixture.roots(), &fixture, &[]),
+            Err(DiscoveryError::InvalidSwapDevice(path)) if path == "/missing.swap"
+        ));
+    }
+
+    #[test]
+    fn partition_holders_exclude_the_namespace() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme0n1", id(259, 0), 8, 0);
+        let partition = fixture.add_partition("nvme0n1", 1, id(259, 1));
+        fixture.add_holder(&partition, "md0");
+
+        let report = discover_with(fixture.roots(), &fixture, &[]).expect("discover disks");
+
+        assert!(report.eligible.is_empty());
+        assert_eq!(report.excluded[0].reason, ExclusionReason::Held);
+    }
+
+    #[test]
+    fn unreadable_partition_identity_excludes_the_namespace() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme0n1", id(259, 0), 8, 0);
+        let partition = fixture.add_partition("nvme0n1", 1, id(259, 1));
+        fs::remove_file(
+            fixture
+                .root
+                .path()
+                .join(format!("sys/class/block/{partition}/dev")),
+        )
+        .expect("remove partition identity");
+
+        let report = discover_with(fixture.roots(), &fixture, &[]).expect("discover disks");
+
+        assert!(report.eligible.is_empty());
+        assert_eq!(
+            report.excluded[0].reason,
+            ExclusionReason::SafetyStateUnavailable
+        );
+    }
+
+    #[test]
+    fn fails_closed_when_usage_state_cannot_be_read() {
+        let mut fixture = Fixture::new();
+        fixture.add_namespace("nvme0n1", id(259, 0), 8, 0);
+        fs::remove_file(fixture.root.path().join("proc/self/mountinfo")).expect("remove mountinfo");
+
+        assert!(matches!(
+            discover_with(fixture.roots(), &fixture, &[]),
+            Err(DiscoveryError::ReadMountInfo(_))
+        ));
+    }
+
+    #[test]
+    fn materializer_preserves_explicit_disks_without_scanning() {
+        let mut config = default_config();
+        config.disks.push(block_spec("/dev/nvme9n1", Some(9)));
+        config.disk_discovery.as_mut().unwrap().denied_paths = vec!["/dev/nvme9n1".into()];
+        let mut scans = 0;
+        let mut resolver = DiskResolver::default();
+
+        let resolution = resolver
+            .resolve_with(&mut config, |_| {
+                scans += 1;
+                Ok(DiscoveryReport::default())
+            })
+            .expect("resolve explicit disks");
+
+        assert_eq!(resolution.source, DiskSource::Explicit);
+        assert_eq!(scans, 0);
+        assert_eq!(config.disks, vec![block_spec("/dev/nvme9n1", Some(9))]);
+    }
+
+    #[test]
+    fn materializer_uses_all_discovered_disks_or_fallback() {
+        let mut resolver = DiskResolver::default();
+        let mut auto_config = default_config();
+        let report = DiscoveryReport {
+            eligible: vec![
+                AutoDisk {
+                    path: "/dev/nvme0n1".into(),
+                    numa: Some(2),
+                },
+                AutoDisk {
+                    path: "/dev/nvme1n1".into(),
+                    numa: None,
+                },
+            ],
+            excluded: Vec::new(),
+        };
+
+        let resolution = resolver
+            .resolve_with(&mut auto_config, |_| Ok(report.clone()))
+            .expect("resolve automatic disks");
+
+        assert_eq!(resolution.source, DiskSource::Automatic);
+        assert_eq!(resolution.report, Some(report));
+        assert_eq!(
+            auto_config.disks,
+            vec![
+                block_spec("/dev/nvme0n1", Some(2)),
+                block_spec("/dev/nvme1n1", None),
+            ]
+        );
+
+        let mut fallback_config = default_config();
+        fallback_config
+            .disk_discovery
+            .as_mut()
+            .unwrap()
+            .fallback
+            .as_mut()
+            .unwrap()
+            .path = "/var/cache/custom.disk".into();
+        let resolution = resolver
+            .resolve_with(&mut fallback_config, |_| Ok(DiscoveryReport::default()))
+            .expect("resolve fallback disk");
+
+        assert_eq!(resolution.source, DiskSource::Fallback);
+        assert_eq!(
+            fallback_config.disks,
+            vec![file_spec("/var/cache/custom.disk", 20 * 1024 * 1024 * 1024)]
+        );
+    }
+
+    #[test]
+    fn materializer_reuses_unchanged_policy_and_rescans_after_explicit_config() {
+        let mut resolver = DiskResolver::default();
+        let scans = Cell::new(0);
+        let mut scan = |_: &[String]| {
+            scans.set(scans.get() + 1);
+            Ok(DiscoveryReport {
+                eligible: vec![AutoDisk {
+                    path: format!("/dev/nvme{}n1", scans.get()),
+                    numa: None,
+                }],
+                excluded: Vec::new(),
+            })
+        };
+
+        let mut first = default_config();
+        resolver.resolve_with(&mut first, &mut scan).unwrap();
+        let mut unchanged = default_config();
+        resolver.resolve_with(&mut unchanged, &mut scan).unwrap();
+        assert_eq!(scans.get(), 1);
+        assert_eq!(unchanged.disks, first.disks);
+
+        let mut changed = default_config();
+        changed.disk_discovery.as_mut().unwrap().denied_paths = vec!["/dev/nvme0n1".into()];
+        resolver.resolve_with(&mut changed, &mut scan).unwrap();
+        assert_eq!(scans.get(), 2);
+
+        let mut explicit = default_config();
+        explicit.disks.push(block_spec("/dev/nvme9n1", None));
+        resolver.resolve_with(&mut explicit, &mut scan).unwrap();
+        assert_eq!(scans.get(), 2);
+
+        let mut auto_again = default_config();
+        resolver.resolve_with(&mut auto_again, &mut scan).unwrap();
+        assert_eq!(scans.get(), 3);
+    }
+
+    fn default_config() -> crate::config::Config {
+        let mut config = crate::config::Config::default();
+        config.apply_defaults();
+        config
+    }
+
+    fn block_spec(path: &str, numa: Option<u32>) -> crate::config::DiskSpec {
+        crate::config::DiskSpec {
+            config: Some(crate::config::disk_spec::Config::Block(
+                crate::config::BlockDiskConfig {
+                    numa,
+                    path: path.into(),
+                },
+            )),
+            ..Default::default()
+        }
+    }
+
+    fn file_spec(path: &str, size: u64) -> crate::config::DiskSpec {
+        crate::config::DiskSpec {
+            config: Some(crate::config::disk_spec::Config::File(
+                crate::config::FileDiskConfig {
+                    path: path.into(),
+                    size: Some(size),
+                },
+            )),
+            ..Default::default()
+        }
+    }
+}
diff --git a/cmd/unbounded-storage/src/lib.rs b/cmd/unbounded-storage/src/lib.rs
index 08c48e5c0..f203369ba 100644
--- a/cmd/unbounded-storage/src/lib.rs
+++ b/cmd/unbounded-storage/src/lib.rs
@@ -4,6 +4,7 @@
 pub mod backend;
 pub mod bufferpool;
 pub mod config;
+pub mod disk_discovery;
 pub mod fabric;
 pub mod fanout;
 pub mod frontend;
diff --git a/cmd/unbounded-storage/src/main.rs b/cmd/unbounded-storage/src/main.rs
index c59658480..bdc016934 100644
--- a/cmd/unbounded-storage/src/main.rs
+++ b/cmd/unbounded-storage/src/main.rs
@@ -18,7 +18,11 @@ use clap::Parser;
 use unbounded_storage::backend::BackendRegistry;
 use unbounded_storage::bufferpool::{Pool, PoolConfig};
 use unbounded_storage::config::{
-    self, BackendSpec, Config, FrontendSpec, LoadedConfig, ResolvedFrontendBinding, frontend_spec,
+    self, BackendSpec, Config, FrontendSpec, LoadedConfig, ResolvedFrontendBinding, disk_spec,
+    frontend_spec,
+};
+use unbounded_storage::disk_discovery::{
+    DiscoveryError, DiscoveryReport, DiskResolution, DiskResolver, DiskSource,
 };
 use unbounded_storage::fabric::{self, Fabric, MrHandle, Provider};
 use unbounded_storage::fanout::{
@@ -226,6 +230,15 @@ fn main() -> ExitCode {
             return ExitCode::FAILURE;
         }
     };
+    let mut disk_resolver = DiskResolver::default();
+    let (loaded, disk_resolution) = match resolve_loaded_config(&loaded, &mut disk_resolver) {
+        Ok(resolved) => resolved,
+        Err(e) => {
+            eprintln!("disk discovery error: {e}");
+            return ExitCode::FAILURE;
+        }
+    };
+    log_disk_resolution(&disk_resolution);
 
     // Startup-fixed settings come from the config file's `[startup]`
     // section, not the dynamically reloaded sections. They size the
@@ -297,7 +310,9 @@ fn main() -> ExitCode {
         core_plan_config,
     });
 
-    let core_plan = CorePlan::for_host(&host, &settings.core_plan_config);
+    let storage_numa = block_disk_numa_hints(loaded.config());
+    let core_plan =
+        CorePlan::for_host_with_storage(&host, &settings.core_plan_config, &storage_numa);
 
     let nic_worker_cpus: usize = core_plan
         .nic_workers
@@ -386,15 +401,11 @@ fn main() -> ExitCode {
     };
 
     // Disk supervisor: reconcile projected cache disks onto pinned
-    // storage cores. Each disk runs on its own storage core hosting the
-    // engine and ring, and publishes a `PageChannel` that carries the
-    // page data path cross-core from the shards. CPU pin hints come from
-    // the topology plan's per-drive storage cores: each one already
-    // inherits the drive's NUMA node and is disjoint from the serving
-    // and NIC-worker cores by construction. If the host discovered no
-    // NVMe devices the slot list is empty and disks run unpinned. The
-    // registry is owned by the apply target so live cache disk changes
-    // reconcile through the same funnel as the rest of the config.
+    // storage cores. Initial block disks receive slots using their NUMA
+    // hints; file disks and block disks added by a later reload run
+    // unpinned. The registry is owned by the apply target so live cache
+    // disk changes reconcile through the same funnel as the rest of the
+    // config.
     let disk_slots: Vec<DiskCpuSlot> = core_plan
         .storage_cores
         .iter()
@@ -497,8 +508,22 @@ fn main() -> ExitCode {
             while !SHUTDOWN.load(Ordering::Acquire) {
                 match update_rx.recv_timeout(SHUTDOWN_POLL) {
                     Ok(update) => {
-                        let version = update.loaded.config().version;
-                        match controller.apply(update.loaded.clone()) {
+                        let (loaded, resolution) = match resolve_loaded_config(
+                            update.loaded.as_ref(),
+                            &mut disk_resolver,
+                        ) {
+                            Ok(resolved) => resolved,
+                            Err(e) => {
+                                eprintln!(
+                                    "config: disk resolution gen={} failed; keeping previous: {e}",
+                                    update.generation
+                                );
+                                continue;
+                            }
+                        };
+                        log_disk_resolution(&resolution);
+                        let version = loaded.config().version;
+                        match controller.apply(Arc::new(loaded)) {
                             Ok(outcome) => eprintln!(
                                 "config: applied gen={} version={} tier={:?}",
                                 update.generation, version, outcome.tier
@@ -1659,6 +1684,59 @@ fn load_config(path: &Path, explicit: bool) -> Result<LoadedConfig, String> {
     }
 }
 
+fn resolve_loaded_config(
+    loaded: &LoadedConfig,
+    resolver: &mut DiskResolver,
+) -> Result<(LoadedConfig, DiskResolution), String> {
+    resolve_loaded_config_with(
+        loaded,
+        resolver,
+        unbounded_storage::disk_discovery::discover,
+    )
+}
+
+fn resolve_loaded_config_with(
+    loaded: &LoadedConfig,
+    resolver: &mut DiskResolver,
+    scan: impl FnMut(&[String]) -> Result<DiscoveryReport, DiscoveryError>,
+) -> Result<(LoadedConfig, DiskResolution), String> {
+    let mut config = loaded.config().clone();
+    let resolution = resolver
+        .resolve_with(&mut config, scan)
+        .map_err(|e| e.to_string())?;
+    let loaded = LoadedConfig::from_config(config).map_err(|e| e.to_string())?;
+    Ok((loaded, resolution))
+}
+
+fn block_disk_numa_hints(config: &Config) -> Vec<Option<u16>> {
+    config
+        .disks
+        .iter()
+        .filter_map(|disk| match disk.config.as_ref() {
+            Some(disk_spec::Config::Block(block)) => {
+                Some(block.numa.and_then(|numa| u16::try_from(numa).ok()))
+            }
+            Some(disk_spec::Config::File(_)) | None => None,
+        })
+        .collect()
+}
+
+fn log_disk_resolution(resolution: &DiskResolution) {
+    match resolution.source {
+        DiskSource::Explicit => eprintln!("disk config: using explicit disks"),
+        DiskSource::Automatic => {
+            let count = resolution
+                .report
+                .as_ref()
+                .map_or(0, |report| report.eligible.len());
+            eprintln!("disk config: automatically selected {count} NVMe namespace(s)");
+        }
+        DiskSource::Fallback => {
+            eprintln!("disk config: no eligible NVMe namespaces; using fallback file")
+        }
+    }
+}
+
 /// Block the calling thread until the process-wide [`SHUTDOWN`] latch is
 /// set, polling on [`SHUTDOWN_POLL`]. Used to park the main thread while
 /// the shard threads run.
@@ -2015,6 +2093,30 @@ mod tests {
         assert!(load_config(path, true).is_err());
     }
 
+    #[test]
+    fn loaded_config_is_resolved_before_runtime_projection() {
+        let loaded = LoadedConfig::from_config(Config::default()).unwrap();
+        let mut resolver = unbounded_storage::disk_discovery::DiskResolver::default();
+
+        let (resolved, resolution) = resolve_loaded_config_with(&loaded, &mut resolver, |_| {
+            Ok(unbounded_storage::disk_discovery::DiscoveryReport {
+                eligible: vec![unbounded_storage::disk_discovery::AutoDisk {
+                    path: "/dev/nvme0n1".into(),
+                    numa: Some(3),
+                }],
+                excluded: Vec::new(),
+            })
+        })
+        .expect("resolve loaded config");
+
+        assert_eq!(
+            resolution.source,
+            unbounded_storage::disk_discovery::DiskSource::Automatic
+        );
+        assert_eq!(resolved.config().disks.len(), 1);
+        assert_eq!(config::runtime_disks(resolved.runtime()).len(), 1);
+    }
+
     fn backend_spec(id: &str) -> BackendSpec {
         BackendSpec {
             name: id.to_string(),
diff --git a/cmd/unbounded-storage/src/topology/cores.rs b/cmd/unbounded-storage/src/topology/cores.rs
index a8282f359..3d39f4b6e 100644
--- a/cmd/unbounded-storage/src/topology/cores.rs
+++ b/cmd/unbounded-storage/src/topology/cores.rs
@@ -7,8 +7,8 @@
 //! by partitioning the host's usable CPUs into three pinned classes,
 //! scheduled in order of how constrained they are:
 //!
-//! 1. **Storage cores** (most constrained): exactly one CPU per NVMe
-//!    drive, preferring a CPU local to the drive's `(numa, pcie_root)`.
+//! 1. **Storage cores** (most constrained): exactly one CPU per concrete
+//!    block disk, preferring a CPU local to the disk's NUMA node.
 //! 2. **NIC workers**: `nic_workers` CPUs per active HCA, preferring
 //!    CPUs that share the HCA's `(numa, pcie_root)`, round-robin within
 //!    the local pool.
@@ -35,8 +35,9 @@ use std::collections::{BTreeMap, VecDeque};
 use super::Host;
 use super::filters::{self, Filters};
 
-/// A CPU dedicated to one NVMe drive's storage engine and io_uring
-/// ring. `drive` indexes into `host.nvmes`.
+/// A CPU dedicated to one block disk's storage engine and io_uring
+/// ring. `drive` indexes into the concrete block-disk list supplied to
+/// [`CorePlan::for_host_with_storage`].
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct StorageCore {
     pub cpu: u32,
@@ -77,8 +78,8 @@ pub struct NumaPool {
     pub workers: usize,
 }
 
-/// A CPU slot reserved for a disk storage core: one per disk path (per
-/// NVMe drive), drawn from the plan's [`StorageCore`] allocation. These
+/// A CPU slot reserved for a disk storage core: one per concrete block
+/// disk, drawn from the plan's [`StorageCore`] allocation. These
 /// slots come out of the same disjoint, NUMA-local, SMT-collapsed,
 /// cpu0-excluded allocator that places the serving shards and NIC
 /// workers, so a disk pinned to one of these CPUs never collides with a
@@ -157,20 +158,32 @@ impl CorePlanConfig {
 }
 
 impl CorePlan {
-    /// Build a three-class core plan for `host` under `cfg`. The result
-    /// is deterministic: NVMes and HCAs are consumed in the sorted order
-    /// `Host::discover` produced, and CPUs come out of each per-NUMA
-    /// pool in ascending id order.
+    /// Build a three-class core plan using the NVMe controllers discovered
+    /// on `host`. This compatibility entry point preserves the original
+    /// controller-based behavior; runtime callers with concrete disk paths
+    /// should use [`Self::for_host_with_storage`].
     pub fn for_host(host: &Host, cfg: &CorePlanConfig) -> Self {
+        let storage_numa = host.nvmes.iter().map(|nvme| nvme.numa).collect::<Vec<_>>();
+        Self::for_host_with_storage(host, cfg, &storage_numa)
+    }
+
+    /// Build a plan with one storage core per concrete block disk. File
+    /// disks are omitted by the caller because they do not have device
+    /// locality and can run without a dedicated pinned storage slot.
+    pub fn for_host_with_storage(
+        host: &Host,
+        cfg: &CorePlanConfig,
+        storage_numa: &[Option<u16>],
+    ) -> Self {
         let filters = cfg.filters();
         let kept_hcas = filters::filter_hcas(&host.hcas, &filters);
         let mut alloc = CoreAllocator::new(host, &filters);
 
         // 1. Storage cores first (most constrained): one CPU per drive,
         //    local to the drive's NUMA node when possible.
-        let mut storage_cores = Vec::with_capacity(host.nvmes.len());
-        for (drive, nvme) in host.nvmes.iter().enumerate() {
-            let (cpu, numa) = alloc.take(nvme.numa);
+        let mut storage_cores = Vec::with_capacity(storage_numa.len());
+        for (drive, preferred_numa) in storage_numa.iter().copied().enumerate() {
+            let (cpu, numa) = alloc.take(preferred_numa);
             storage_cores.push(StorageCore { cpu, numa, drive });
         }
 
@@ -564,6 +577,28 @@ mod tests {
         assert!(!set(cpus).contains(&0));
     }
 
+    #[test]
+    fn storage_cores_follow_concrete_block_disks_not_nvme_controllers() {
+        let host = fake_host(
+            vec![
+                (0, (0..8).collect(), vec![]),
+                (1, (8..16).collect(), vec![]),
+            ],
+            vec![],
+            vec![],
+            vec![nvme("nvme0", "0000:02:00.0", Some(0))],
+        );
+
+        let plan = CorePlan::for_host_with_storage(&host, &defaults(), &[Some(1), Some(1), None]);
+
+        assert_eq!(plan.storage_cores.len(), 3);
+        assert_eq!(plan.storage_cores[0].drive, 0);
+        assert_eq!(plan.storage_cores[0].numa, Some(1));
+        assert_eq!(plan.storage_cores[1].drive, 1);
+        assert_eq!(plan.storage_cores[1].numa, Some(1));
+        assert_eq!(plan.storage_cores[2].drive, 2);
+    }
+
     #[test]
     fn isolcpus_restricts_pool() {
         let host = fake_host(
diff --git a/deploy/unbounded-storage-supervisor/02-configmap.yaml.tmpl b/deploy/unbounded-storage-supervisor/02-configmap.yaml.tmpl
index 44de434d1..f58d6db6a 100644
--- a/deploy/unbounded-storage-supervisor/02-configmap.yaml.tmpl
+++ b/deploy/unbounded-storage-supervisor/02-configmap.yaml.tmpl
@@ -25,18 +25,21 @@ metadata:
 # deployment should participate in P2P caching; the supervisor only supplies the
 # ring roster.
 #
-# Disks are declared in disks[]. Explicit disks are preserved. When disks[] is
-# absent or empty, block disks are allocated only from the opt-in
-# storage.unbounded-cloud.io/allocated-disks annotation. Its value is a
-# comma-separated list of item?query entries such as
-# /dev/nvme1n1?queue_depth=256&page_size_bytes=4096&numa=0. When that annotation
-# is absent or has no valid disks, the supervisor uses one file-backed disk at
-# /var/lib/unbounded-storage/cache.disk sized by
-# unbounded-cloud.io/storage-file-size-bytes, defaulting to 2 GiB. The daemon
-# publishes discovered options to storage.unbounded-cloud.io/block-devices as
-# entries like /dev/sdb?name=sdb&size_bytes=1234; discovery does not allocate
-# disks. The shared disk set is available to every cache; the supervisor does not
-# invent caches, backends, or frontends.
+# Disks are declared in disks[]. Explicit disks are authoritative. The existing
+# storage.unbounded-cloud.io/allocated-disks annotation is also preserved as an
+# explicit per-node override; its value is a comma-separated list of item?query
+# entries such as /dev/nvme1n1?queue_depth=256&page_size_bytes=4096&numa=0.
+# Otherwise the daemon automatically selects whole NVMe namespaces that are not
+# mounted, active swap, or held by LVM/RAID/device-mapper (including their child
+# partitions). Exact device paths can be excluded with
+# disk_discovery.denied_paths. If no eligible namespace remains, the daemon uses
+# disk_discovery.fallback, defaulting to /var/lib/unbounded-storage/cache.disk at
+# 20 GiB. The legacy unbounded-cloud.io/storage-file-size-bytes annotation sets
+# the fallback size only when config.yaml did not set it. The daemon also
+# publishes inventory to storage.unbounded-cloud.io/block-devices; inventory is
+# informational and includes devices that are not eligible. The shared disk set
+# is available to every cache; the supervisor does not invent caches, backends,
+# or frontends.
 #
 # Backend ca_cert, client_cert, and client_key fields contain literal PEM, not
 # filesystem paths. Static S3 access_key_id, secret_access_key, and session_token
@@ -47,6 +50,16 @@ data:
     # Operator-assigned config version (opaque counter; 0 means unversioned).
     version: 0
 
+    # Automatic disk policy used only when disks[] and the allocated-disks node
+    # annotation are empty. Denied paths are exact and do not filter explicit
+    # disks. Omit fallback fields for the default path and 20 GiB size.
+    # disk_discovery:
+    #   denied_paths:
+    #     - /dev/nvme0n1
+    #   fallback:
+    #     path: /var/lib/unbounded-storage/cache.disk
+    #     size: 21474836480
+
     # Optional automatic P2P finger weighting. Omit topology_weighting to keep
     # historical hard-locality selection. When set, weight 0.0 is neutral,
     # positive values favor matching peers, and negative values penalize them.
diff --git a/internal/storagesupervisor/disks.go b/internal/storagesupervisor/disks.go
index 88fa0c861..152745ddf 100644
--- a/internal/storagesupervisor/disks.go
+++ b/internal/storagesupervisor/disks.go
@@ -16,9 +16,7 @@ import (
 const (
 	allocatedDisksAnnotation   = "storage.unbounded-cloud.io/allocated-disks"
 	storageFileSizeAnnotation  = "unbounded-cloud.io/storage-file-size-bytes"
-	defaultStorageFileDiskPath = "/var/lib/unbounded-storage/cache.disk"
 	defaultStorageFileDiskDir  = "/var/lib/unbounded-storage"
-	defaultStorageFileDiskSize = uint64(2 * 1024 * 1024 * 1024)
 	defaultStorageDiskPageSize = 4096
 	diskOptionQueueDepth       = "queue_depth"
 	diskOptionPageSizeBytes    = "page_size_bytes"
@@ -30,32 +28,51 @@ const (
 	diskOptionNuma             = "numa"
 )
 
-// applyDiskOverlay injects per-node storage disks when the config does not
-// declare disks explicitly. Explicit config disks are authoritative.
+// applyDiskOverlay preserves explicit config and annotation disks. When neither
+// is present, the daemon discovers safe NVMe namespaces itself. The legacy file
+// size annotation only customizes that discovery policy's fallback file.
 func applyDiskOverlay(cfg *storageconfig.Config, annotations map[string]string) error {
 	if len(cfg.GetDisks()) > 0 {
 		return nil
 	}
 
-	existingPaths := declaredDiskPaths(cfg)
+	disks := annotationBlockDisks(annotations[allocatedDisksAnnotation], map[string]struct{}{})
+	if len(disks) > 0 {
+		cfg.Disks = disks
 
-	disks := annotationBlockDisks(annotations[allocatedDisksAnnotation], existingPaths)
-	if len(disks) == 0 {
-		fallback := fallbackFileDisk(annotations[storageFileSizeAnnotation])
+		return nil
+	}
 
-		path := fallback.GetFile().GetPath()
-		if _, exists := existingPaths[path]; exists {
-			slog.Warn("skipping fallback storage file disk because path is already declared", "path", path)
+	applyFallbackSizeOverlay(cfg, annotations[storageFileSizeAnnotation])
 
-			return nil
-		}
+	return nil
+}
 
-		disks = []*storageconfig.DiskSpec{fallback}
+func applyFallbackSizeOverlay(cfg *storageconfig.Config, rawSize string) {
+	rawSize = strings.TrimSpace(rawSize)
+	if rawSize == "" {
+		return
 	}
 
-	cfg.Disks = disks
+	parsed, err := strconv.ParseUint(rawSize, 10, 64)
+	if err != nil || parsed == 0 || parsed%defaultStorageDiskPageSize != 0 {
+		slog.Warn("ignoring invalid storage fallback file size annotation",
+			"annotation", storageFileSizeAnnotation, "value", rawSize)
 
-	return nil
+		return
+	}
+
+	if cfg.DiskDiscovery == nil {
+		cfg.DiskDiscovery = &storageconfig.DiskDiscoveryCfg{}
+	}
+
+	if cfg.DiskDiscovery.Fallback == nil {
+		cfg.DiskDiscovery.Fallback = &storageconfig.FileDiskConfig{}
+	}
+
+	if cfg.DiskDiscovery.Fallback.Size == nil {
+		cfg.DiskDiscovery.Fallback.Size = proto.Uint64(parsed)
+	}
 }
 
 func annotationBlockDisks(raw string, existingPaths map[string]struct{}) []*storageconfig.DiskSpec {
@@ -302,55 +319,3 @@ func parsePositiveUint64Option(path, key, raw string) (uint64, bool) {
 
 	return v, true
 }
-
-func fallbackFileDisk(rawSize string) *storageconfig.DiskSpec {
-	size := defaultStorageFileDiskSize
-
-	if strings.TrimSpace(rawSize) != "" {
-		parsed, err := strconv.ParseUint(strings.TrimSpace(rawSize), 10, 64)
-		if err != nil || parsed == 0 || parsed%defaultStorageDiskPageSize != 0 {
-			slog.Warn("using default storage file disk size because annotation is invalid",
-				"annotation", storageFileSizeAnnotation, "value", rawSize, "default", defaultStorageFileDiskSize)
-		} else {
-			size = parsed
-		}
-	}
-
-	return &storageconfig.DiskSpec{
-		Config: &storageconfig.DiskSpec_File{
-			File: &storageconfig.FileDiskConfig{
-				Path: defaultStorageFileDiskPath,
-				Size: proto.Uint64(size),
-			},
-		},
-	}
-}
-
-func declaredDiskPaths(cfg *storageconfig.Config) map[string]struct{} {
-	paths := map[string]struct{}{}
-
-	for _, disk := range cfg.GetDisks() {
-		path := diskPath(disk)
-		if path != "" {
-			paths[path] = struct{}{}
-		}
-	}
-
-	return paths
-}
-
-func diskPath(disk *storageconfig.DiskSpec) string {
-	if disk == nil {
-		return ""
-	}
-
-	if block := disk.GetBlock(); block != nil {
-		return block.GetPath()
-	}
-
-	if file := disk.GetFile(); file != nil {
-		return file.GetPath()
-	}
-
-	return ""
-}
diff --git a/internal/storagesupervisor/render.go b/internal/storagesupervisor/render.go
index 4ce6b0756..9135b6386 100644
--- a/internal/storagesupervisor/render.go
+++ b/internal/storagesupervisor/render.go
@@ -37,10 +37,10 @@ type renderState struct {
 // computed from the Kubernetes node watch. When the ring is active, this node's
 // peer name is injected and discovered peers are merged with any peers declared
 // in the YAML (discovered peers win on name collision). TCP rings also override
-// startup.fabric.tcp.addr with the node's own routable bind. The default disk
-// set is populated from the self node's storage disk annotations, or from a
-// default file-backed disk when no valid annotation disks are present. Loadgen
-// annotations append synthetic frontends for this node.
+// startup.fabric.tcp.addr with the node's own routable bind. Storage disk
+// annotations remain authoritative when present; otherwise an empty disk list
+// reaches the daemon for safe NVMe discovery. Loadgen annotations append
+// synthetic frontends for this node.
 func RenderConfig(sourceDir string, state renderState) ([]byte, error) {
 	cfg, err := loadSourceConfig(sourceDir)
 	if err != nil {
diff --git a/internal/storagesupervisor/render_test.go b/internal/storagesupervisor/render_test.go
index ed4556cf1..3d682ec18 100644
--- a/internal/storagesupervisor/render_test.go
+++ b/internal/storagesupervisor/render_test.go
@@ -669,7 +669,7 @@ version: 1
 	assert.Equal(t, "/dev/nvme1n1", disk.GetBlock().GetPath())
 }
 
-func TestRenderConfigAllInvalidAnnotatedDisksFallsBackToFile(t *testing.T) {
+func TestRenderConfigAllInvalidAnnotatedDisksLeavesDiscoveryToDaemon(t *testing.T) {
 	dir := writeSource(t, `
 version: 1
 `)
@@ -681,14 +681,14 @@ version: 1
 		},
 	})
 
-	require.Len(t, cfg.GetDisks(), 1)
-	disk := cfg.GetDisks()[0]
-	require.NotNil(t, disk.GetFile())
-	assert.Equal(t, defaultStorageFileDiskPath, disk.GetFile().GetPath())
-	assert.Equal(t, uint64(4294967296), disk.GetFile().GetSize())
+	assert.Empty(t, cfg.GetDisks())
+	require.NotNil(t, cfg.GetDiskDiscovery())
+	require.NotNil(t, cfg.GetDiskDiscovery().GetFallback())
+	assert.Empty(t, cfg.GetDiskDiscovery().GetFallback().GetPath())
+	assert.Equal(t, uint64(4294967296), cfg.GetDiskDiscovery().GetFallback().GetSize())
 }
 
-func TestRenderConfigBlankAnnotatedDisksFallsBackToFile(t *testing.T) {
+func TestRenderConfigBlankAnnotatedDisksLeavesDiscoveryToDaemon(t *testing.T) {
 	dir := writeSource(t, `
 version: 1
 `)
@@ -700,27 +700,22 @@ version: 1
 		},
 	})
 
-	require.Len(t, cfg.GetDisks(), 1)
-	disk := cfg.GetDisks()[0]
-	assert.Equal(t, defaultStorageFileDiskPath, disk.GetFile().GetPath())
-	assert.Equal(t, uint64(4294967296), disk.GetFile().GetSize())
+	assert.Empty(t, cfg.GetDisks())
+	assert.Equal(t, uint64(4294967296), cfg.GetDiskDiscovery().GetFallback().GetSize())
 }
 
-func TestRenderConfigFallbackFileDiskDefaultSize(t *testing.T) {
+func TestRenderConfigWithoutDiskAnnotationsLeavesDiscoveryToDaemon(t *testing.T) {
 	dir := writeSource(t, `
 version: 1
 `)
 
 	cfg := decode(t, dir)
 
-	require.Len(t, cfg.GetDisks(), 1)
-	disk := cfg.GetDisks()[0]
-	assert.Equal(t, defaultStorageFileDiskPath, disk.GetFile().GetPath())
-	assert.Equal(t, defaultStorageFileDiskSize, disk.GetFile().GetSize())
-	assert.False(t, disk.GetSkipRecoveryScan())
+	assert.Empty(t, cfg.GetDisks())
+	assert.Nil(t, cfg.GetDiskDiscovery())
 }
 
-func TestRenderConfigInvalidFileSizeAnnotationFallsBackToDefault(t *testing.T) {
+func TestRenderConfigInvalidFileSizeAnnotationLeavesDaemonDefaults(t *testing.T) {
 	tests := []struct {
 		name string
 		size string
@@ -740,12 +735,45 @@ version: 1
 				annotations: map[string]string{storageFileSizeAnnotation: tt.size},
 			})
 
-			require.Len(t, cfg.GetDisks(), 1)
-			assert.Equal(t, defaultStorageFileDiskSize, cfg.GetDisks()[0].GetFile().GetSize())
+			assert.Empty(t, cfg.GetDisks())
+			assert.Nil(t, cfg.GetDiskDiscovery())
 		})
 	}
 }
 
+func TestRenderConfigFileSizeAnnotationDoesNotOverrideConfiguredFallback(t *testing.T) {
+	dir := writeSource(t, `
+disk_discovery:
+  fallback:
+    path: /custom/fallback.disk
+    size: 1073741824
+`)
+
+	cfg := decodeWithState(t, dir, renderState{
+		annotations: map[string]string{storageFileSizeAnnotation: "4294967296"},
+	})
+
+	assert.Empty(t, cfg.GetDisks())
+	assert.Equal(t, "/custom/fallback.disk", cfg.GetDiskDiscovery().GetFallback().GetPath())
+	assert.Equal(t, uint64(1073741824), cfg.GetDiskDiscovery().GetFallback().GetSize())
+}
+
+func TestRenderConfigFileSizeAnnotationPreservesConfiguredFallbackPath(t *testing.T) {
+	dir := writeSource(t, `
+disk_discovery:
+  fallback:
+    path: /custom/fallback.disk
+`)
+
+	cfg := decodeWithState(t, dir, renderState{
+		annotations: map[string]string{storageFileSizeAnnotation: "4294967296"},
+	})
+
+	assert.Empty(t, cfg.GetDisks())
+	assert.Equal(t, "/custom/fallback.disk", cfg.GetDiskDiscovery().GetFallback().GetPath())
+	assert.Equal(t, uint64(4294967296), cfg.GetDiskDiscovery().GetFallback().GetSize())
+}
+
 func TestRenderConfigPreservesExplicitConfigMapDisks(t *testing.T) {
 	dir := writeSource(t, `
 disks:
@@ -781,7 +809,7 @@ caches:
 	require.Len(t, cfg.GetCaches(), 2)
 	assert.Equal(t, "cache-a", cfg.GetCaches()[0].GetName())
 	assert.Equal(t, "cache-b", cfg.GetCaches()[1].GetName())
-	require.Len(t, cfg.GetDisks(), 1)
+	assert.Empty(t, cfg.GetDisks())
 }
 
 func TestRenderConfigNoCachesInjectsDisks(t *testing.T) {