diff --git a/crates/clickhousectl/src/error.rs b/crates/clickhousectl/src/error.rs
index 9406f350..2f84b1dc 100644
--- a/crates/clickhousectl/src/error.rs
+++ b/crates/clickhousectl/src/error.rs
@@ -254,6 +254,62 @@ pub enum Error {
     #[error("Server '{0}' is running; stop it first with `clickhousectl local server stop {0}`")]
     ServerRunningCannotRemove(String),
 
+    #[error(
+        "Permission denied reading server metadata '{}': {source}. Check ownership and file permissions, then retry.",
+        path.display()
+    )]
+    ServerMetadataPermission {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Failed to read server metadata '{}': {source}. Check that the file is readable, then retry.",
+        path.display()
+    )]
+    ServerMetadataRead {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error(
+        "Server metadata '{}' is not valid UTF-8: {source}. Repair or remove the metadata file, then retry.",
+        path.display()
+    )]
+    ServerMetadataUtf8 {
+        path: PathBuf,
+        #[source]
+        source: std::string::FromUtf8Error,
+    },
+
+    #[error(
+        "Server metadata '{}' is not valid JSON: {source}. Repair or remove the metadata file, then retry.",
+        path.display()
+    )]
+    ServerMetadataParse {
+        path: PathBuf,
+        #[source]
+        source: serde_json::Error,
+    },
+
+    #[error("Failed to durably update server metadata '{}': {source}", path.display())]
+    ServerMetadataWrite {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error("Could not {operation} at '{}': {source}. {remediation}", path.display())]
+    ServerLock {
+        operation: &'static str,
+        path: PathBuf,
+        remediation: &'static str,
+        #[source]
+        source: std::io::Error,
+    },
+
     #[error("{0}")]
     Cloud(String),
 
diff --git a/crates/clickhousectl/src/local/docker.rs b/crates/clickhousectl/src/local/docker.rs
index fe9b5653..f990274e 100644
--- a/crates/clickhousectl/src/local/docker.rs
+++ b/crates/clickhousectl/src/local/docker.rs
@@ -1085,13 +1085,16 @@ pub async fn start_existing(docker: &Docker, id: &str) -> Result<()> {
 /// Safe to call multiple times in one CLI invocation. When Docker isn't
 /// reachable, `connect()` fails fast (no socket → immediate error, no I/O
 /// timeout) and we return silently.
-pub fn recover_project_postgres_blocking(project_cwd: &str) {
+pub fn recover_project_postgres_blocking(
+    project_cwd: &str,
+    lock: &crate::local::server::MetadataLock,
+) -> Result<()> {
     use crate::local::server::{
-        Engine, ServerInfo, ensure_pg_data_dir, pg_instance_key, save_server_info,
-        server_meta_path_for_recovery,
+        Engine, ServerInfo, ensure_pg_data_dir, load_info_locked, pg_instance_key,
+        save_server_info_locked,
     };
     let cwd_owned = project_cwd.to_string();
-    let _ = block_on(async move {
+    block_on(async move {
         let docker = match connect().await {
             Ok(d) => d,
             Err(_) => return Ok::<(), Error>(()),
@@ -1102,10 +1105,10 @@ pub fn recover_project_postgres_blocking(project_cwd: &str) {
         };
         for c in containers {
             let key = pg_instance_key(&c.user_name, &c.major);
-            if server_meta_path_for_recovery(&key).exists() {
+            if load_info_locked(&key, lock)?.is_some() {
                 continue;
             }
-            let _ = ensure_pg_data_dir(&c.user_name, &c.major);
+            ensure_pg_data_dir(&c.user_name, &c.major)?;
             let info = ServerInfo {
                 name: key,
                 pid: 0,
@@ -1117,10 +1120,10 @@ pub fn recover_project_postgres_blocking(project_cwd: &str) {
                 engine: Engine::Postgres,
                 container_id: Some(c.container_id.clone()),
             };
-            let _ = save_server_info(&info);
+            save_server_info_locked(&info, lock)?;
         }
         Ok(())
-    });
+    })
 }
 
 #[cfg(test)]
diff --git a/crates/clickhousectl/src/local/mod.rs b/crates/clickhousectl/src/local/mod.rs
index 93be4ba1..692a9d58 100644
--- a/crates/clickhousectl/src/local/mod.rs
+++ b/crates/clickhousectl/src/local/mod.rs
@@ -190,8 +190,8 @@ fn remove(version: &str, force: bool, json: bool) -> Result<()> {
     // Recover orphaned servers so we detect a running process even when its
     // metadata file is missing, then refuse to pull the binary out from under
     // a server running on this version.
-    server::recover_current_project_servers();
-    let in_use: Vec<String> = server::list_running_servers()
+    server::recover_current_project_servers()?;
+    let in_use: Vec<String> = server::list_running_servers()?
         .into_iter()
         .filter(|i| i.version == version)
         .map(|i| i.name)
@@ -274,19 +274,17 @@ fn run_client(
         (h, p, v)
     } else {
         let server_name = name.as_deref().unwrap_or("default");
-        let entries = server::list_all_servers();
-        let entry = entries
-            .iter()
-            .find(|e| e.name == server_name)
+        let metadata_lock = server::lock_metadata()?;
+        server::recover_current_project_servers_locked(&metadata_lock)?;
+        let entry = server::server_entry_locked(server_name, &metadata_lock)?
             .ok_or_else(|| Error::ServerNotFound(server_name.to_string()))?;
         if !entry.running {
             return Err(Error::ServerNotRunning(server_name.to_string()));
         }
         let info = entry
             .info
-            .as_ref()
             .ok_or_else(|| Error::ServerNotRunning(server_name.to_string()))?;
-        ("localhost".to_string(), info.tcp_port, info.version.clone())
+        ("localhost".to_string(), info.tcp_port, info.version)
     };
 
     let binary = paths::binary_path(&version)?;
@@ -358,6 +356,21 @@ fn resolve_direct_client_version(version_spec: Option<ClientVersionArg>) -> Resu
     }
 }
 
+fn clean_up_untracked_child(child: &mut std::process::Child, primary: Error) -> Error {
+    let cleanup = match child.try_wait() {
+        Ok(Some(_)) => Ok(()),
+        Ok(None) => child.kill().and_then(|()| child.wait()).map(|_| ()),
+        Err(error) => Err(error),
+    };
+    match cleanup {
+        Ok(()) => primary,
+        Err(error) => Error::Exec(format!(
+            "{primary}; additionally failed to stop untracked PID {}: {error}",
+            child.id()
+        )),
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn start_server(
     name: Option<String>,
@@ -375,12 +388,12 @@ async fn start_server(
 
     // Recover any orphaned servers so name resolution and collision checks
     // see processes that lost their metadata files.
-    server::recover_current_project_servers();
+    server::recover_current_project_servers()?;
 
     // Resolve server name and check for collisions before any downloads
     let server_name = server::resolve_name(name.as_deref())?;
 
-    if name.is_some() && server::is_server_running(&server_name) {
+    if name.is_some() && server::is_server_running(&server_name)? {
         return Err(Error::ServerAlreadyRunning(server_name));
     }
 
@@ -416,8 +429,18 @@ async fn start_server(
         return Err(Error::VersionNotFound(version));
     }
 
+    // Metadata locking is always acquired after version installation locks.
+    // Hold it from the final collision check through the process metadata
+    // commit so concurrent start/stop/client commands see one lifecycle state.
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
+    let server_name = server::resolve_name_locked(name.as_deref(), &metadata_lock)?;
+    if name.is_some() && server::is_server_running_locked(&server_name, &metadata_lock)? {
+        return Err(Error::ServerAlreadyRunning(server_name));
+    }
+
     // Show running server count
-    let running = server::running_server_count();
+    let running = server::advisory_running_server_count_locked(&metadata_lock);
     if running > 0 {
         eprintln!(
             "Note: {} server{} already running (use `clickhousectl local server list` to see them)",
@@ -509,7 +532,10 @@ async fn start_server(
             engine: server::Engine::Clickhouse,
             container_id: None,
         };
-        server::save_server_info(&info)?;
+        if let Err(error) = server::save_server_info_locked(&info, &metadata_lock) {
+            return Err(clean_up_untracked_child(&mut child, error));
+        }
+        drop(metadata_lock);
 
         if no_wait {
             server::check_spawn_health(&mut child, &server_name, &log_path).await?;
@@ -549,7 +575,10 @@ async fn start_server(
             engine: server::Engine::Clickhouse,
             container_id: None,
         };
-        server::save_server_info(&info)?;
+        if let Err(error) = server::save_server_info_locked(&info, &metadata_lock) {
+            return Err(clean_up_untracked_child(&mut child, error));
+        }
+        drop(metadata_lock);
 
         eprintln!(
             "Server '{}' running (PID: {}, HTTP: {}, TCP: {})",
@@ -587,17 +616,15 @@ fn dotenv_server(
     json: bool,
 ) -> Result<()> {
     let server_name = name.unwrap_or("default");
-    let entries = server::list_all_servers();
-    let entry = entries
-        .iter()
-        .find(|e| e.name == server_name)
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
+    let entry = server::server_entry_locked(server_name, &metadata_lock)?
         .ok_or_else(|| Error::ServerNotFound(server_name.to_string()))?;
     if !entry.running {
         return Err(Error::ServerNotRunning(server_name.to_string()));
     }
     let info = entry
         .info
-        .as_ref()
         .ok_or_else(|| Error::ServerNotRunning(server_name.to_string()))?;
 
     // Only write vars we actually know from server metadata.
@@ -769,17 +796,18 @@ async fn run_server_commands(command: ServerCommands, json: bool) -> Result<()>
 
                 // Recover orphaned servers so we can stop processes
                 // that lost their metadata files.
-                server::recover_current_project_servers();
+                let metadata_lock = server::lock_metadata()?;
+                server::recover_current_project_servers_locked(&metadata_lock)?;
 
                 match classify_stop(
-                    server::is_server_running(&name),
+                    server::is_server_running_locked(&name, &metadata_lock)?,
                     server::server_data_dir(&name).exists(),
                 ) {
                     StopOutcome::Stop => {
                         if !json {
                             println!("Stopping server '{}'...", name);
                         }
-                        server::kill_server(&name)?;
+                        server::kill_server_locked(&name, &metadata_lock)?;
                         let out = output::ServerStopOutput {
                             name,
                             already_stopped: false,
@@ -822,9 +850,10 @@ async fn run_server_commands(command: ServerCommands, json: bool) -> Result<()>
 
             // Recover orphaned servers so we correctly detect a running
             // process even when its metadata file is missing.
-            server::recover_current_project_servers();
+            let metadata_lock = server::lock_metadata()?;
+            server::recover_current_project_servers_locked(&metadata_lock)?;
 
-            if server::is_server_running(&name) {
+            if server::is_server_running_locked(&name, &metadata_lock)? {
                 return Err(Error::ServerRunningCannotRemove(name));
             }
             let data_dir = server::server_data_dir(&name);
@@ -834,7 +863,7 @@ async fn run_server_commands(command: ServerCommands, json: bool) -> Result<()>
             // Remove the whole server directory (parent of data/)
             let server_dir = data_dir.parent().unwrap();
             std::fs::remove_dir_all(server_dir)?;
-            server::remove_server_info(&name);
+            server::try_remove_server_info_locked(&name, &metadata_lock)?;
             let out = output::ServerRemoveOutput { name };
             output::print_output(&out, json);
             Ok(())
@@ -863,7 +892,7 @@ fn classify_stop(running: bool, exists_on_disk: bool) -> StopOutcome {
 }
 
 fn list_servers_local(json: bool) -> Result<()> {
-    let entries = server::list_all_servers();
+    let entries = server::list_all_servers()?;
     let running_count = entries.iter().filter(|e| e.running).count();
     let total = entries.len();
 
@@ -1006,8 +1035,12 @@ fn stop_server_global(name: &str, project: Option<&str>, json: bool) -> Result<(
 }
 
 fn stop_all_servers_local(json: bool) -> Result<()> {
-    let servers = server::list_running_servers();
-    let out = stop_servers(&servers, json, server::kill_server);
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
+    let servers = server::list_running_servers_locked(&metadata_lock)?;
+    let out = stop_servers(&servers, json, |name| {
+        server::kill_server_locked(name, &metadata_lock)
+    });
     if json {
         output::print_output(&out, json);
     } else if servers.is_empty() {
diff --git a/crates/clickhousectl/src/local/postgres.rs b/crates/clickhousectl/src/local/postgres.rs
index a71dfd94..21b279c0 100644
--- a/crates/clickhousectl/src/local/postgres.rs
+++ b/crates/clickhousectl/src/local/postgres.rs
@@ -248,69 +248,63 @@ async fn start(
     let extra_env = preflight.extra_env;
     let password_from_env = preflight.password_from_env;
 
-    server::recover_current_project_servers();
-
-    // User-facing name (no version suffix). Defaults to "default" when no
-    // postgres "default" is currently running.
-    let user_name = match name.as_deref() {
-        Some(n) => n.to_string(),
-        None => default_pg_name(),
-    };
-
-    // If `--version` is omitted but there's already exactly one instance for
-    // this name, resume it — the user almost certainly wants their existing
-    // data, not a freshly-initialized DEFAULT_PG_TAG. With multiple
-    // instances, we ask them to disambiguate. Only when zero exist do we
-    // default to DEFAULT_PG_TAG.
-    let (tag, major) = match version.as_deref() {
-        Some(v) => (v.to_string(), pg_major_from_tag(v)),
-        None => {
-            let existing = server::find_pg_instances(&user_name);
-            match existing.len() {
-                0 => (
-                    DEFAULT_PG_TAG.to_string(),
-                    pg_major_from_tag(DEFAULT_PG_TAG),
-                ),
-                1 => {
-                    let info = &existing[0];
-                    let stored_tag = info
-                        .version
-                        .strip_prefix("postgres:")
-                        .unwrap_or(&info.version);
-                    (stored_tag.to_string(), pg_major_from_tag(stored_tag))
-                }
-                _ => {
-                    let versions: Vec<String> =
-                        existing.iter().map(|i| i.version.clone()).collect();
-                    return Err(Error::Postgres(format!(
-                        "multiple postgres instances named '{}' ({}); pass --version to select one",
-                        user_name,
-                        versions.join(", ")
-                    )));
-                }
-            }
-        }
-    };
-    let tag = tag.as_str();
-
-    // Disk identifier — uniquely scopes (name, major) so two majors of the
-    // same name never share container/data/metadata.
-    let key = server::pg_instance_key(&user_name, &major);
-
     let docker = docker::connect().await?;
-
     let project_cwd = std::env::current_dir()
         .and_then(|p| p.canonicalize())
         .map(|p| p.display().to_string())
         .unwrap_or_default();
 
-    // Resume path: an instance for this exact (name, major) already exists.
-    if let Some(prior) = server::load_info(&key) {
-        let cid = prior.container_id.as_deref().unwrap_or("");
-        let container_present =
-            !cid.is_empty() && docker.inspect_container(cid, None).await.is_ok();
-        if container_present {
-            if server::is_server_running(&key) {
+    loop {
+        // Resolve an optimistic target, then release the project-wide lock
+        // before potentially slow fresh-image inspection and pulling.
+        let metadata_lock = server::lock_metadata()?;
+        server::recover_current_project_servers_locked(&metadata_lock)?;
+        let user_name = match name.as_deref() {
+            Some(name) => name.to_string(),
+            None => default_pg_name_locked(&metadata_lock)?,
+        };
+        let (tag, major) =
+            resolve_pg_start_version_locked(&user_name, version.as_deref(), &metadata_lock)?;
+        let key = server::pg_instance_key(&user_name, &major);
+        let prior = server::load_info_locked(&key, &metadata_lock)?;
+        drop(metadata_lock);
+
+        if prior.is_none() {
+            if !docker::image_exists(&docker, &tag).await? {
+                docker::pull_image(&docker, &tag, json).await?;
+            }
+            docker::ensure_name_free(&docker, &user_name, &major, &project_cwd).await?;
+        }
+
+        // The optimistic target may have changed while Docker work was in
+        // progress. Re-resolve it before any state-determining mutation.
+        let metadata_lock = server::lock_metadata()?;
+        let (current_tag, current_major) =
+            resolve_pg_start_version_locked(&user_name, version.as_deref(), &metadata_lock)?;
+        let current_key = server::pg_instance_key(&user_name, &current_major);
+        let current = server::load_info_locked(&current_key, &metadata_lock)?;
+        if current_tag != tag || current_key != key || current != prior {
+            drop(metadata_lock);
+            continue;
+        }
+
+        // Resume path: an instance for this exact (name, major) already exists.
+        if let Some(prior) = prior {
+            let cid = prior.container_id.as_deref().unwrap_or("");
+            let inspected = if cid.is_empty() {
+                None
+            } else {
+                docker.inspect_container(cid, None).await.ok()
+            };
+            let Some(inspected) = inspected else {
+                return Err(Error::Postgres(format!(
+                    "server '{}' (postgres:{}) has metadata but the container is gone. \
+                     Run `clickhousectl local postgres remove {}` to clear the data dir \
+                     and start fresh.",
+                    user_name, major, user_name
+                )));
+            };
+            if inspected.state.and_then(|state| state.running) == Some(true) {
                 return Err(Error::ServerAlreadyRunning(user_name));
             }
             if port.is_some()
@@ -325,108 +319,100 @@ async fn start(
                     user_name, user_name
                 );
             }
-            return resume_existing(&docker, prior, wait_timeout, json).await;
+            return resume_existing(&docker, prior, wait_timeout, json, metadata_lock).await;
         }
-        // Metadata orphaned — container removed externally. Force explicit
-        // cleanup to avoid silently re-initing against potentially-stale data.
-        return Err(Error::Postgres(format!(
-            "server '{}' (postgres:{}) has metadata but the container is gone. \
-             Run `clickhousectl local postgres remove {}` to clear the data dir \
-             and start fresh.",
-            user_name, major, user_name
-        )));
-    }
-
-    // Fresh create.
-    let host_port = match host_port {
-        Some(port) => port,
-        None => resolve_port(None)?,
-    };
-    if !docker::image_exists(&docker, tag).await? {
-        docker::pull_image(&docker, tag, json).await?;
-    }
-
-    let instance_dir = server::servers_dir_join(&key);
-    let remove_fresh_data_on_failure = fresh_instance_dir_is_disposable(&instance_dir);
-    server::ensure_pg_data_dir(&user_name, &major)?;
-    let data_dir = server::pg_data_dir(&user_name, &major);
-
-    // Defensive cleanup of any unmanaged container colliding on our chosen
-    // container name (only if labels confirm we own it).
-    docker::ensure_name_free(&docker, &user_name, &major, &project_cwd).await?;
 
-    let user = user.unwrap_or_else(|| DEFAULT_USER.to_string());
-    let database = database.unwrap_or_else(|| DEFAULT_DATABASE.to_string());
+        // Fresh create.
+        let host_port = match host_port {
+            Some(port) => port,
+            None => resolve_port(None)?,
+        };
 
-    let password = password_from_env
-        .or(password)
-        .unwrap_or_else(generate_password);
+        let instance_dir = server::servers_dir_join(&key);
+        let remove_fresh_data_on_failure = fresh_instance_dir_is_disposable(&instance_dir);
+        server::ensure_pg_data_dir(&user_name, &major)?;
+        let data_dir = server::pg_data_dir(&user_name, &major);
+
+        let user = user.unwrap_or_else(|| DEFAULT_USER.to_string());
+        let database = database.unwrap_or_else(|| DEFAULT_DATABASE.to_string());
+
+        let password = password_from_env
+            .or(password)
+            .unwrap_or_else(generate_password);
+
+        let opts = PostgresRunOpts {
+            user_name: &user_name,
+            major: &major,
+            tag: &tag,
+            host_port,
+            data_dir: &data_dir,
+            project_cwd: &project_cwd,
+            user: &user,
+            password: &password,
+            database: &database,
+            extra_env,
+        };
 
-    let opts = PostgresRunOpts {
-        user_name: &user_name,
-        major: &major,
-        tag,
-        host_port,
-        data_dir: &data_dir,
-        project_cwd: &project_cwd,
-        user: &user,
-        password: &password,
-        database: &database,
-        extra_env,
-    };
+        let container_id = docker::create_postgres(&docker, opts).await?;
+
+        let info = ServerInfo {
+            name: key.clone(),
+            pid: 0,
+            version: format!("postgres:{tag}"),
+            http_port: 0,
+            tcp_port: host_port,
+            started_at: server::now_timestamp(),
+            cwd: project_cwd.clone(),
+            engine: Engine::Postgres,
+            container_id: Some(container_id.clone()),
+        };
+        let startup_result = async {
+            docker::start_existing(&docker, &container_id).await?;
+            server::save_server_info_locked(&info, &metadata_lock)
+        }
+        .await;
 
-    let container_id = docker::create_postgres(&docker, opts).await?;
+        if let Err(primary) = startup_result {
+            return Err(rollback_failed_fresh_start(
+                &docker,
+                &container_id,
+                &info,
+                remove_fresh_data_on_failure,
+                primary,
+                &metadata_lock,
+            )
+            .await);
+        }
+        drop(metadata_lock);
 
-    let info = ServerInfo {
-        name: key.clone(),
-        pid: 0,
-        version: format!("postgres:{tag}"),
-        http_port: 0,
-        tcp_port: host_port,
-        started_at: server::now_timestamp(),
-        cwd: project_cwd.clone(),
-        engine: Engine::Postgres,
-        container_id: Some(container_id.clone()),
-    };
-    let startup_result = async {
-        docker::start_existing(&docker, &container_id).await?;
-        server::save_server_info(&info)?;
         if let Err(failure) = wait_for_postgres_ready(&docker, &container_id, wait_timeout).await {
-            return Err(postgres_readiness_error(
+            let primary =
+                postgres_readiness_error(&docker, &container_id, &user_name, wait_timeout, failure)
+                    .await;
+            let metadata_lock = server::lock_metadata()?;
+            return Err(rollback_failed_fresh_start(
                 &docker,
                 &container_id,
-                &user_name,
-                wait_timeout,
-                failure,
+                &info,
+                remove_fresh_data_on_failure,
+                primary,
+                &metadata_lock,
             )
             .await);
         }
-        Ok(())
-    }
-    .await;
 
-    if let Err(primary) = startup_result {
-        return Err(rollback_failed_fresh_start(
-            &docker,
-            &container_id,
-            &info,
-            remove_fresh_data_on_failure,
-            primary,
-        )
-        .await);
+        let out = output::PostgresStartOutput {
+            name: user_name,
+            container_id,
+            image: format!("postgres:{tag}"),
+            port: host_port,
+            user,
+            password,
+            database,
+        };
+        output::print_output(&out, json);
+        return Ok(());
     }
-
-    let out = output::PostgresStartOutput {
-        name: user_name,
-        container_id,
-        image: format!("postgres:{tag}"),
-        port: host_port,
-        user,
-        password,
-        database,
-    };
-    output::print_output(&out, json);
-    Ok(())
 }
 
 /// A fresh attempt owns an absent or empty instance directory, including one
@@ -460,9 +446,10 @@ async fn rollback_failed_fresh_start(
     info: &ServerInfo,
     remove_fresh_data_on_failure: bool,
     primary: Error,
+    metadata_lock: &server::MetadataLock,
 ) -> Error {
     let instance_dir = server::servers_dir_join(&info.name);
-    let metadata_path = server::server_meta_path_for_recovery(&info.name);
+    let metadata_path = server::servers_dir_join(&format!("{}.json", info.name));
     let mut diagnostics = Vec::new();
 
     let container_removed = match docker::remove_container(docker, container_id).await {
@@ -507,16 +494,15 @@ async fn rollback_failed_fresh_start(
     };
 
     if container_removed && instance_removed {
-        match std::fs::remove_file(&metadata_path) {
+        match server::try_remove_server_info_locked(&info.name, metadata_lock) {
             Ok(()) => return primary,
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return primary,
             Err(error) => diagnostics.push(format!(
                 "failed to remove metadata '{}': {error}",
                 metadata_path.display()
             )),
         }
     } else {
-        match server::save_server_info(info) {
+        match server::save_server_info_locked(info, metadata_lock) {
             Ok(()) => diagnostics.push(format!(
                 "recovery metadata retained at '{}'; run `clickhousectl local postgres remove {}` to clean up",
                 metadata_path.display(),
@@ -537,35 +523,76 @@ async fn rollback_failed_fresh_start(
 
 /// Default user-facing name when `--name` is omitted: `"default"` if no
 /// postgres "default" is running, otherwise a random adjective-noun.
-fn default_pg_name() -> String {
-    let any_default_running = server::find_pg_instances("default").iter().any(|i| {
-        i.container_id
-            .as_deref()
-            .map(docker::is_container_running_blocking)
-            .unwrap_or(false)
-    });
+fn default_pg_name_locked(metadata_lock: &server::MetadataLock) -> Result<String> {
+    let any_default_running = server::find_pg_instances_locked("default", metadata_lock)?
+        .iter()
+        .any(|i| {
+            i.container_id
+                .as_deref()
+                .map(docker::is_container_running_blocking)
+                .unwrap_or(false)
+        });
     if any_default_running {
         // Fall back to the existing random-name generator, which checks
         // metadata file uniqueness across engines.
-        server::resolve_name(None).unwrap_or_else(|_| "default".into())
+        server::resolve_name_locked(None, metadata_lock)
     } else {
-        "default".into()
+        Ok("default".into())
+    }
+}
+
+/// Resolve the image tag for a user-facing name. The caller invokes this both
+/// before Docker image preparation and under the final metadata lock.
+fn resolve_pg_start_version_locked(
+    user_name: &str,
+    version: Option<&str>,
+    metadata_lock: &server::MetadataLock,
+) -> Result<(String, String)> {
+    if let Some(version) = version {
+        return Ok((version.to_string(), pg_major_from_tag(version)));
+    }
+
+    let existing = server::find_pg_instances_locked(user_name, metadata_lock)?;
+    match existing.as_slice() {
+        [] => Ok((
+            DEFAULT_PG_TAG.to_string(),
+            pg_major_from_tag(DEFAULT_PG_TAG),
+        )),
+        [info] => {
+            let stored_tag = info
+                .version
+                .strip_prefix("postgres:")
+                .unwrap_or(&info.version);
+            Ok((stored_tag.to_string(), pg_major_from_tag(stored_tag)))
+        }
+        _ => {
+            let versions: Vec<&str> = existing.iter().map(|info| info.version.as_str()).collect();
+            Err(Error::Postgres(format!(
+                "multiple postgres instances named '{}' ({}); pass --version to select one",
+                user_name,
+                versions.join(", ")
+            )))
+        }
     }
 }
 
 /// Resolve `--name <X> [--version <V>]` to a single Postgres instance on disk.
 /// If `version` is given, target the (X, major(V)) pair directly. Otherwise:
 /// 0 instances → ServerNotFound; 1 → use it; >1 → ask for `--version`.
-fn resolve_pg_target(user_name: &str, version: Option<&str>) -> Result<server::ServerInfo> {
+fn resolve_pg_target_locked(
+    user_name: &str,
+    version: Option<&str>,
+    metadata_lock: &server::MetadataLock,
+) -> Result<server::ServerInfo> {
     if let Some(v) = version {
         validate_pg_tag(v)?;
         let major = pg_major_from_tag(v);
         let key = server::pg_instance_key(user_name, &major);
-        return server::load_info(&key)
+        return server::load_info_locked(&key, metadata_lock)?
             .filter(|i| i.engine == Engine::Postgres)
             .ok_or_else(|| Error::ServerNotFound(format!("{user_name} (postgres:{major})")));
     }
-    let instances = server::find_pg_instances(user_name);
+    let instances = server::find_pg_instances_locked(user_name, metadata_lock)?;
     match instances.len() {
         0 => Err(Error::ServerNotFound(user_name.to_string())),
         1 => Ok(instances.into_iter().next().unwrap()),
@@ -588,12 +615,31 @@ async fn resume_existing(
     prior: ServerInfo,
     wait_timeout: Duration,
     json: bool,
+    metadata_lock: server::MetadataLock,
 ) -> Result<()> {
     let container_id = prior.container_id.clone().expect("checked by caller");
     let display_name = user_name_from_key(&prior.name).to_string();
 
     docker::start_existing(docker, &container_id).await?;
 
+    let info = ServerInfo {
+        started_at: server::now_timestamp(),
+        ..prior
+    };
+    if let Err(primary) = server::save_server_info_locked(&info, &metadata_lock) {
+        drop(metadata_lock);
+        return match docker::stop_container(docker, &container_id).await {
+            Ok(()) => Err(primary),
+            Err(cleanup) => Err(Error::PostgresStartupRollback {
+                primary: Box::new(primary),
+                cleanup: format!(
+                    "could not stop resumed container '{container_id}' after metadata failure: {cleanup}"
+                ),
+            }),
+        };
+    }
+    drop(metadata_lock);
+
     if let Err(failure) = wait_for_postgres_ready(docker, &container_id, wait_timeout).await {
         let error =
             postgres_readiness_error(docker, &container_id, &display_name, wait_timeout, failure)
@@ -604,12 +650,6 @@ async fn resume_existing(
 
     let (user, password, database) = read_pg_env(docker, &container_id).await;
 
-    let info = ServerInfo {
-        started_at: server::now_timestamp(),
-        ..prior
-    };
-    server::save_server_info(&info)?;
-
     let out = output::PostgresStartOutput {
         name: display_name,
         container_id,
@@ -902,13 +942,14 @@ fn generate_password() -> String {
 
 async fn stop(name: &str, version: Option<&str>, json: bool) -> Result<()> {
     server::validate_server_name(name)?;
-    server::recover_current_project_servers();
-    let target = resolve_pg_target(name, version)?;
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
+    let target = resolve_pg_target_locked(name, version, &metadata_lock)?;
     if !json {
         let display = format!("{} ({})", user_name_from_key(&target.name), target.version);
         println!("Stopping Postgres {}...", display);
     }
-    server::kill_server(&target.name)?;
+    server::kill_server_locked(&target.name, &metadata_lock)?;
     let out = output::ServerStopOutput {
         name: user_name_from_key(&target.name).to_string(),
         already_stopped: false,
@@ -918,8 +959,9 @@ async fn stop(name: &str, version: Option<&str>, json: bool) -> Result<()> {
 }
 
 async fn stop_all(json: bool) -> Result<()> {
-    server::recover_current_project_servers();
-    let servers: Vec<_> = server::list_running_servers()
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
+    let servers: Vec<_> = server::list_running_servers_locked(&metadata_lock)?
         .into_iter()
         .filter(|s| s.engine == Engine::Postgres)
         .collect();
@@ -928,7 +970,9 @@ async fn stop_all(json: bool) -> Result<()> {
         return Ok(());
     }
 
-    let out = super::stop_servers(&servers, json, server::kill_server);
+    let out = super::stop_servers(&servers, json, |name| {
+        server::kill_server_locked(name, &metadata_lock)
+    });
     if json {
         output::print_output(&out, json);
     } else {
@@ -939,11 +983,12 @@ async fn stop_all(json: bool) -> Result<()> {
 
 fn remove(name: &str, version: Option<&str>, json: bool) -> Result<()> {
     server::validate_server_name(name)?;
-    server::recover_current_project_servers();
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
 
-    let target = resolve_pg_target(name, version)?;
+    let target = resolve_pg_target_locked(name, version, &metadata_lock)?;
     let key = target.name.clone();
-    if server::is_server_running(&key) {
+    if server::is_server_running_locked(&key, &metadata_lock)? {
         return Err(Error::ServerAlreadyRunning(name.to_string()));
     }
 
@@ -958,7 +1003,7 @@ fn remove(name: &str, version: Option<&str>, json: bool) -> Result<()> {
     // privileged container in that case.
     let pg_dir = server::servers_dir_join(&key);
     docker::remove_host_dir_blocking(&pg_dir)?;
-    server::remove_server_info(&key);
+    server::try_remove_server_info_locked(&key, &metadata_lock)?;
     let out = output::ServerRemoveOutput {
         name: name.to_string(),
     };
@@ -992,12 +1037,14 @@ async fn client(
         );
     }
 
-    server::recover_current_project_servers();
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
     let server_name = name.as_deref().unwrap_or("default");
-    let info = resolve_pg_target(server_name, version.as_deref())?;
-    if !server::is_server_running(&info.name) {
+    let info = resolve_pg_target_locked(server_name, version.as_deref(), &metadata_lock)?;
+    if !server::is_server_running_locked(&info.name, &metadata_lock)? {
         return Err(Error::ServerNotRunning(server_name.to_string()));
     }
+    drop(metadata_lock);
 
     let docker = docker::connect().await?;
     let container_id = info
@@ -1110,12 +1157,14 @@ fn exec_host_psql(
 }
 
 fn dotenv(name: Option<&str>, version: Option<&str>, use_local: bool, json: bool) -> Result<()> {
-    server::recover_current_project_servers();
+    let metadata_lock = server::lock_metadata()?;
+    server::recover_current_project_servers_locked(&metadata_lock)?;
     let server_name = name.unwrap_or("default");
-    let info = resolve_pg_target(server_name, version)?;
-    if !server::is_server_running(&info.name) {
+    let info = resolve_pg_target_locked(server_name, version, &metadata_lock)?;
+    if !server::is_server_running_locked(&info.name, &metadata_lock)? {
         return Err(Error::ServerNotRunning(server_name.to_string()));
     }
+    drop(metadata_lock);
 
     // Read user/password/db from the container env so we always emit accurate creds.
     let (user, password, database) = docker::block_on(read_pg_env_for_dotenv(
diff --git a/crates/clickhousectl/src/local/server.rs b/crates/clickhousectl/src/local/server.rs
index a79e4e5c..7d7517ed 100644
--- a/crates/clickhousectl/src/local/server.rs
+++ b/crates/clickhousectl/src/local/server.rs
@@ -3,6 +3,8 @@ use crate::init;
 use crate::local::discovery;
 use crate::local::docker;
 use serde::{Deserialize, Serialize};
+use std::fs::{File, OpenOptions};
+use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -13,6 +15,8 @@ const SPAWN_HEALTH_DELAY: Duration = Duration::from_millis(300);
 const STARTUP_POLL_INTERVAL: Duration = Duration::from_millis(100);
 const CONNECT_TIMEOUT: Duration = Duration::from_millis(200);
 const STARTUP_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(3);
+const METADATA_LOCK_FILE: &str = ".metadata.lock";
+const METADATA_TEMP_PREFIX: &str = ".metadata-";
 
 const ADJECTIVES: &[&str] = &[
     "bold", "calm", "dark", "fast", "gold", "keen", "loud", "neat", "pale", "red", "slim", "tall",
@@ -51,7 +55,7 @@ fn default_engine() -> Engine {
 /// `engine` and `container_id` are post-Postgres-support additions and default
 /// to ClickHouse + None so existing `.clickhouse/servers/*.json` files keep
 /// deserializing.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ServerInfo {
     pub name: String,
     /// Active ClickHouse process PID; 0 when stopped or for Postgres.
@@ -100,14 +104,57 @@ fn servers_dir() -> PathBuf {
     init::local_dir().join("servers")
 }
 
-fn server_meta_path(name: &str) -> PathBuf {
-    servers_dir().join(format!("{}.json", name))
+/// The one project-wide metadata lock. Lifecycle operations hold this lock
+/// from their final state read through their state-determining write. No code
+/// holding it may acquire an install lock, and metadata helpers with a
+/// `_locked` suffix never acquire it again.
+pub(crate) struct MetadataLock {
+    _file: File,
+    dir: PathBuf,
 }
 
-/// Public alias used by `docker.rs` during orphan recovery — keeps the path
-/// computation in one place.
-pub fn server_meta_path_for_recovery(name: &str) -> PathBuf {
-    server_meta_path(name)
+impl MetadataLock {
+    fn acquire_at(dir: &Path) -> Result<Self> {
+        std::fs::create_dir_all(dir).map_err(|source| {
+            server_lock_error(
+                "create the server metadata lock directory",
+                dir,
+                "Check write access to the parent directory, then retry.",
+                source,
+            )
+        })?;
+        let path = dir.join(METADATA_LOCK_FILE);
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(&path)
+            .map_err(|source| {
+                server_lock_error(
+                    "open the server metadata lock file",
+                    &path,
+                    "Check read and write access to the lock file and its directory, then retry.",
+                    source,
+                )
+            })?;
+        file.lock().map_err(|source| {
+            server_lock_error(
+                "acquire the server metadata lock",
+                &path,
+                "Check that the filesystem supports advisory file locks, then retry.",
+                source,
+            )
+        })?;
+        Ok(Self {
+            _file: file,
+            dir: dir.to_path_buf(),
+        })
+    }
+}
+
+pub(crate) fn lock_metadata() -> Result<MetadataLock> {
+    MetadataLock::acquire_at(&servers_dir())
 }
 
 /// Data directory for a ClickHouse server: .clickhouse/servers/<name>/data/.
@@ -179,29 +226,77 @@ pub fn ensure_pg_data_dir(name: &str, major: &str) -> Result<bool> {
     Ok(created)
 }
 
-/// Save server info to a metadata file.
-pub fn save_server_info(info: &ServerInfo) -> Result<()> {
-    let dir = servers_dir();
-    std::fs::create_dir_all(&dir)?;
-    let path = server_meta_path(&info.name);
-    let json = serde_json::to_string_pretty(info)?;
-    std::fs::write(path, json)?;
+fn metadata_write_error(path: &Path, source: std::io::Error) -> Error {
+    Error::ServerMetadataWrite {
+        path: path.to_path_buf(),
+        source,
+    }
+}
+
+fn server_lock_error(
+    operation: &'static str,
+    path: &Path,
+    remediation: &'static str,
+    source: std::io::Error,
+) -> Error {
+    Error::ServerLock {
+        operation,
+        path: path.to_path_buf(),
+        remediation,
+        source,
+    }
+}
+
+fn sync_directory(dir: &Path, metadata_path: &Path) -> Result<()> {
+    #[cfg(unix)]
+    File::open(dir)
+        .and_then(|directory| directory.sync_all())
+        .map_err(|error| metadata_write_error(metadata_path, error))?;
     Ok(())
 }
 
-/// Remove a server's metadata file.
-pub fn remove_server_info(name: &str) {
-    let _ = try_remove_server_info(name);
+fn save_server_info_at(dir: &Path, info: &ServerInfo) -> Result<()> {
+    save_server_info_at_with_sync(dir, info, sync_directory)
 }
 
-/// Remove a server's metadata file while retaining cleanup errors for callers
-/// that are rolling back a transaction.
-pub fn try_remove_server_info(name: &str) -> Result<()> {
-    match std::fs::remove_file(server_meta_path(name)) {
-        Ok(()) => Ok(()),
-        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
-        Err(error) => Err(error.into()),
+fn save_server_info_at_with_sync(
+    dir: &Path,
+    info: &ServerInfo,
+    sync: impl FnOnce(&Path, &Path) -> Result<()>,
+) -> Result<()> {
+    let path = dir.join(format!("{}.json", info.name));
+    let json = serde_json::to_vec_pretty(info)?;
+    let mut temporary = tempfile::Builder::new()
+        .prefix(METADATA_TEMP_PREFIX)
+        .tempfile_in(dir)
+        .map_err(|error| metadata_write_error(&path, error))?;
+    temporary
+        .write_all(&json)
+        .and_then(|()| temporary.flush())
+        .and_then(|()| temporary.as_file().sync_all())
+        .map_err(|error| metadata_write_error(&path, error))?;
+    temporary
+        .persist(&path)
+        .map_err(|error| metadata_write_error(&path, error.error))?;
+    // The rename has committed metadata at this point. A directory sync error
+    // must not make callers treat the child as untracked and terminate it.
+    let _ = sync(dir, &path);
+    Ok(())
+}
+
+pub(crate) fn save_server_info_locked(info: &ServerInfo, lock: &MetadataLock) -> Result<()> {
+    validate_server_name(&info.name)?;
+    save_server_info_at(&lock.dir, info)
+}
+
+pub(crate) fn try_remove_server_info_locked(name: &str, lock: &MetadataLock) -> Result<()> {
+    let path = lock.dir.join(format!("{name}.json"));
+    match std::fs::remove_file(&path) {
+        Ok(()) => {}
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+        Err(error) => return Err(metadata_write_error(&path, error)),
     }
+    sync_directory(&lock.dir, &path)
 }
 
 /// Mark a ClickHouse server as stopped without discarding its metadata.
@@ -209,7 +304,12 @@ pub fn try_remove_server_info(name: &str) -> Result<()> {
 /// The PID match avoids overwriting metadata when a newer process was already
 /// recorded before this transition began.
 pub fn mark_server_stopped(name: &str, pid: u32) -> Result<()> {
-    let Some(mut info) = load_info(name) else {
+    let lock = lock_metadata()?;
+    mark_server_stopped_locked(name, pid, &lock)
+}
+
+pub(crate) fn mark_server_stopped_locked(name: &str, pid: u32, lock: &MetadataLock) -> Result<()> {
+    let Some(mut info) = load_info_locked(name, lock)? else {
         return Ok(());
     };
     if info.engine == Engine::Clickhouse && info.pid == pid {
@@ -217,7 +317,7 @@ pub fn mark_server_stopped(name: &str, pid: u32) -> Result<()> {
         info.version.clear();
         info.http_port = 0;
         info.tcp_port = 0;
-        save_server_info(&info)?;
+        save_server_info_locked(&info, lock)?;
     }
     Ok(())
 }
@@ -233,25 +333,51 @@ fn is_alive(info: &ServerInfo) -> bool {
     }
 }
 
-/// Load server metadata regardless of liveness. Returns None if no metadata
-/// file exists or it can't be parsed. `name` is the disk identifier — for
-/// ClickHouse this is just the user-facing name, for Postgres it's
-/// `<name>-pg<major>` (use `pg_instance_key`).
-pub fn load_info(name: &str) -> Option<ServerInfo> {
-    let content = std::fs::read_to_string(server_meta_path(name)).ok()?;
-    serde_json::from_str(&content).ok()
+fn load_info_at(path: &Path) -> Result<Option<ServerInfo>> {
+    let bytes = match std::fs::read(path) {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(source) if source.kind() == std::io::ErrorKind::PermissionDenied => {
+            return Err(Error::ServerMetadataPermission {
+                path: path.to_path_buf(),
+                source,
+            });
+        }
+        Err(source) => {
+            return Err(Error::ServerMetadataRead {
+                path: path.to_path_buf(),
+                source,
+            });
+        }
+    };
+    let content = String::from_utf8(bytes).map_err(|source| Error::ServerMetadataUtf8 {
+        path: path.to_path_buf(),
+        source,
+    })?;
+    let info = serde_json::from_str(&content).map_err(|source| Error::ServerMetadataParse {
+        path: path.to_path_buf(),
+        source,
+    })?;
+    Ok(Some(info))
+}
+
+pub(crate) fn load_info_locked(name: &str, lock: &MetadataLock) -> Result<Option<ServerInfo>> {
+    validate_server_name(name)?;
+    load_info_at(&lock.dir.join(format!("{name}.json")))
 }
 
 /// Find every Postgres instance whose user-facing name is `name`. Returns
 /// one entry per major version that has a metadata file on disk.
-pub fn find_pg_instances(name: &str) -> Vec<ServerInfo> {
+pub(crate) fn find_pg_instances_locked(name: &str, lock: &MetadataLock) -> Result<Vec<ServerInfo>> {
     let prefix = format!("{}-pg", name);
-    let dir = match std::fs::read_dir(servers_dir()) {
+    let dir = match std::fs::read_dir(&lock.dir) {
         Ok(d) => d,
-        Err(_) => return Vec::new(),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(error) => return Err(error.into()),
     };
     let mut out = Vec::new();
-    for entry in dir.flatten() {
+    for entry in dir {
+        let entry = entry?;
         let fname = match entry.file_name().into_string() {
             Ok(s) => s,
             Err(_) => continue,
@@ -269,22 +395,28 @@ pub fn find_pg_instances(name: &str) -> Vec<ServerInfo> {
         if major.is_empty() || !major.chars().all(|c| c.is_ascii_digit()) {
             continue;
         }
-        if let Some(info) = load_info(stem)
+        if let Some(info) = load_info_locked(stem, lock)?
             && info.engine == Engine::Postgres
         {
             out.push(info);
         }
     }
-    out
+    Ok(out)
 }
 
 /// Load server metadata only if the underlying process/container is alive.
 /// Does not update stale metadata. `list_all_servers` is the single place that
 /// marks ClickHouse entries stopped when their PID is gone, so callers like
 /// `is_server_running` and `resolve_name` can read metadata without side effects.
-fn load_running_info(name: &str) -> Option<ServerInfo> {
-    let info = load_info(name)?;
-    if is_alive(&info) { Some(info) } else { None }
+fn load_running_info_locked(name: &str, lock: &MetadataLock) -> Result<Option<ServerInfo>> {
+    let Some(info) = load_info_locked(name, lock)? else {
+        return Ok(None);
+    };
+    if is_alive(&info) {
+        Ok(Some(info))
+    } else {
+        Ok(None)
+    }
 }
 
 /// List all known servers (both running and stopped).
@@ -293,19 +425,34 @@ fn load_running_info(name: &str) -> Option<ServerInfo> {
 /// entry — for ClickHouse the disk id is the user-facing name; for Postgres
 /// it's `<name>-pg<major>`. Also runs process/container discovery so
 /// orphaned instances reappear.
-pub fn list_all_servers() -> Vec<ServerEntry> {
-    recover_current_project_servers();
+pub fn list_all_servers() -> Result<Vec<ServerEntry>> {
+    let lock = lock_metadata()?;
+    recover_current_project_servers_locked(&lock)?;
+    list_all_servers_locked(&lock)
+}
 
-    let dir = servers_dir();
+pub(crate) fn list_all_servers_locked(lock: &MetadataLock) -> Result<Vec<ServerEntry>> {
+    list_all_servers_locked_inner(lock, false)
+}
+
+fn list_all_servers_locked_inner(
+    lock: &MetadataLock,
+    skip_entry_errors: bool,
+) -> Result<Vec<ServerEntry>> {
+    let dir = &lock.dir;
     let mut entries = Vec::new();
 
-    let dir_entries = match std::fs::read_dir(&dir) {
+    let dir_entries = match std::fs::read_dir(dir) {
         Ok(e) => e,
-        Err(_) => return entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(entries),
+        Err(error) => return Err(error.into()),
     };
 
-    for entry in dir_entries.flatten() {
-        let path = entry.path();
+    for entry in dir_entries {
+        let entry = entry?;
+        if !entry.path().is_file() {
+            continue;
+        }
         let fname = match entry.file_name().into_string() {
             Ok(s) => s,
             Err(_) => continue,
@@ -314,60 +461,86 @@ pub fn list_all_servers() -> Vec<ServerEntry> {
             Some(s) => s,
             None => continue,
         };
-        if !path.is_file() {
+        let entry = match server_entry_locked(stem, lock) {
+            Ok(entry) => entry,
+            Err(_) if skip_entry_errors => continue,
+            Err(error) => return Err(error),
+        };
+        let Some(entry) = entry else {
+            // The file was removed after read_dir; absence is not corruption
+            // and must not produce a phantom stopped entry.
             continue;
-        }
-
-        let mut info = load_info(stem);
-        let mut running = match &info {
-            Some(i) => is_alive(i),
-            None => false,
         };
+        entries.push(entry);
+    }
 
-        // A dead ClickHouse PID can result from a crash or a global stop. Keep
-        // the metadata and normalize it to the same stopped sentinel Postgres
-        // uses so the instance remains discoverable.
-        if let Some(i) = &info
-            && !running
-            && i.engine == Engine::Clickhouse
-            && i.pid != 0
-        {
-            let stale_pid = i.pid;
-            let _ = mark_server_stopped(stem, stale_pid);
-            // Reload because a concurrent restart may have replaced the stale
-            // PID before `mark_server_stopped` acquired the latest metadata.
-            info = load_info(stem);
-            running = info.as_ref().is_some_and(is_alive);
-        }
+    entries.sort_by(|a, b| b.running.cmp(&a.running).then(a.name.cmp(&b.name)));
+    Ok(entries)
+}
 
-        entries.push(ServerEntry {
-            name: stem.to_string(),
-            running,
-            info,
-        });
+pub(crate) fn server_entry_locked(name: &str, lock: &MetadataLock) -> Result<Option<ServerEntry>> {
+    server_entry_locked_with(name, lock, || {})
+}
+
+fn server_entry_locked_with(
+    name: &str,
+    lock: &MetadataLock,
+    before_stale_write: impl FnOnce(),
+) -> Result<Option<ServerEntry>> {
+    let Some(mut info) = load_info_locked(name, lock)? else {
+        return Ok(None);
+    };
+    let mut running = is_alive(&info);
+
+    // Keep the lock across liveness, comparison, and replacement. A restart
+    // either commits before this read or waits and commits after normalization.
+    if !running && info.engine == Engine::Clickhouse && info.pid != 0 {
+        before_stale_write();
+        mark_server_stopped_locked(name, info.pid, lock)?;
+        info =
+            load_info_locked(name, lock)?.ok_or_else(|| Error::ServerNotFound(name.to_string()))?;
+        running = is_alive(&info);
     }
 
-    entries.sort_by(|a, b| b.running.cmp(&a.running).then(a.name.cmp(&b.name)));
-    entries
+    Ok(Some(ServerEntry {
+        name: name.to_string(),
+        running,
+        info: Some(info),
+    }))
 }
 
 /// List only currently running servers.
-pub fn list_running_servers() -> Vec<ServerInfo> {
-    list_all_servers()
+pub fn list_running_servers() -> Result<Vec<ServerInfo>> {
+    Ok(list_all_servers()?
         .into_iter()
         .filter(|e| e.running)
         .filter_map(|e| e.info)
-        .collect()
+        .collect())
+}
+
+pub(crate) fn list_running_servers_locked(lock: &MetadataLock) -> Result<Vec<ServerInfo>> {
+    Ok(list_all_servers_locked(lock)?
+        .into_iter()
+        .filter(|entry| entry.running)
+        .filter_map(|entry| entry.info)
+        .collect())
+}
+
+/// Best-effort count used only for the informational notice during start.
+pub(crate) fn advisory_running_server_count_locked(lock: &MetadataLock) -> usize {
+    list_all_servers_locked_inner(lock, true)
+        .map(|entries| entries.iter().filter(|entry| entry.running).count())
+        .unwrap_or_default()
 }
 
 /// Check if a named server is currently running.
-pub fn is_server_running(name: &str) -> bool {
-    load_running_info(name).is_some()
+pub fn is_server_running(name: &str) -> Result<bool> {
+    let lock = lock_metadata()?;
+    is_server_running_locked(name, &lock)
 }
 
-/// Count running servers.
-pub fn running_server_count() -> usize {
-    list_running_servers().len()
+pub(crate) fn is_server_running_locked(name: &str, lock: &MetadataLock) -> Result<bool> {
+    Ok(load_running_info_locked(name, lock)?.is_some())
 }
 
 fn is_process_alive(pid: u32) -> bool {
@@ -423,12 +596,18 @@ fn kill_process(pid: u32) -> Result<()> {
 ///   the metadata file so a subsequent `start` resumes the same container
 ///   (preserving the password and any other PGDATA-encoded settings).
 pub fn kill_server(name: &str) -> Result<()> {
-    let info = load_running_info(name).ok_or_else(|| Error::ServerNotRunning(name.to_string()))?;
+    let lock = lock_metadata()?;
+    kill_server_locked(name, &lock)
+}
+
+pub(crate) fn kill_server_locked(name: &str, lock: &MetadataLock) -> Result<()> {
+    let info = load_running_info_locked(name, lock)?
+        .ok_or_else(|| Error::ServerNotRunning(name.to_string()))?;
 
     match info.engine {
         Engine::Clickhouse => {
             kill_process(info.pid)?;
-            mark_server_stopped(name, info.pid)?;
+            mark_server_stopped_locked(name, info.pid, lock)?;
         }
         Engine::Postgres => {
             let id = info.container_id.as_deref().ok_or_else(|| {
@@ -448,14 +627,19 @@ pub fn kill_server(name: &str) -> Result<()> {
 /// or generate a random name if "default" is already running.
 /// Returns an error if the provided name contains path traversal characters.
 pub fn resolve_name(name: Option<&str>) -> Result<String> {
+    let lock = lock_metadata()?;
+    resolve_name_locked(name, &lock)
+}
+
+pub(crate) fn resolve_name_locked(name: Option<&str>, lock: &MetadataLock) -> Result<String> {
     match name {
         Some(n) => {
             validate_server_name(n)?;
             Ok(n.to_string())
         }
         None => {
-            if is_server_running("default") {
-                Ok(generate_random_name())
+            if is_server_running_locked("default", lock)? {
+                generate_random_name_locked(lock)
             } else {
                 Ok("default".to_string())
             }
@@ -463,7 +647,7 @@ pub fn resolve_name(name: Option<&str>) -> Result<String> {
     }
 }
 
-fn generate_random_name() -> String {
+fn generate_random_name_locked(lock: &MetadataLock) -> Result<String> {
     let seed = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap_or_default()
@@ -473,15 +657,15 @@ fn generate_random_name() -> String {
     let noun = NOUNS[((mixed / ADJECTIVES.len() as u128) % NOUNS.len() as u128) as usize];
     let tag = format!("{}-{}", adj, noun);
 
-    if is_server_running(&tag) {
+    if is_server_running_locked(&tag, lock)? {
         for i in 2..100 {
             let candidate = format!("{}-{}", tag, i);
-            if !is_server_running(&candidate) {
-                return candidate;
+            if !is_server_running_locked(&candidate, lock)? {
+                return Ok(candidate);
             }
         }
     }
-    tag
+    Ok(tag)
 }
 
 /// Wait a moment after spawn and check if the child exited immediately.
@@ -492,13 +676,18 @@ pub async fn check_spawn_health(
 ) -> Result<()> {
     tokio::time::sleep(SPAWN_HEALTH_DELAY).await;
     if let Some(status) = child.try_wait().map_err(|e| Error::Exec(e.to_string()))? {
-        let _ = mark_server_stopped(name, child.id());
-        return Err(Error::Exec(format!(
+        let message = format!(
             "Server '{}' exited immediately after starting ({}). See server log: {}",
             name,
             status,
             log_path.display()
-        )));
+        );
+        return match mark_server_stopped(name, child.id()) {
+            Ok(()) => Err(Error::Exec(message)),
+            Err(metadata_error) => Err(Error::Exec(format!(
+                "{message}; additionally failed to record the stopped server: {metadata_error}"
+            ))),
+        };
     }
     Ok(())
 }
@@ -551,8 +740,7 @@ pub async fn wait_for_server_ready(
 
     loop {
         if let Some(status) = child.try_wait().map_err(|e| Error::Exec(e.to_string()))? {
-            let _ = mark_server_stopped(name, child.id());
-            return Err(Error::Exec(format!(
+            let message = format!(
                 "Server '{}' exited before becoming ready on HTTP port {} and TCP port {} ({}). \
                  See server log: {}",
                 name,
@@ -560,7 +748,13 @@ pub async fn wait_for_server_ready(
                 tcp_port,
                 status,
                 log_path.display()
-            )));
+            );
+            return match mark_server_stopped(name, child.id()) {
+                Ok(()) => Err(Error::Exec(message)),
+                Err(metadata_error) => Err(Error::Exec(format!(
+                    "{message}; additionally failed to record the stopped server: {metadata_error}"
+                ))),
+            };
         }
 
         let tcp_ready = matches!(
@@ -687,11 +881,16 @@ pub fn now_timestamp() -> String {
 /// `.clickhouse/servers/<name>/data/` path. If a process is found that has no
 /// metadata file, a new `ServerInfo` is saved so it appears in `server list`
 /// and can be managed normally.
-pub fn recover_current_project_servers() {
-    let current_dir = match std::env::current_dir().and_then(|p| p.canonicalize()) {
-        Ok(p) => p.display().to_string(),
-        Err(_) => return,
-    };
+pub fn recover_current_project_servers() -> Result<()> {
+    let lock = lock_metadata()?;
+    recover_current_project_servers_locked(&lock)
+}
+
+pub(crate) fn recover_current_project_servers_locked(lock: &MetadataLock) -> Result<()> {
+    let current_dir = std::env::current_dir()?
+        .canonicalize()?
+        .display()
+        .to_string();
 
     let processes = discovery::discover_clickhouse_processes();
     for proc in processes {
@@ -705,11 +904,7 @@ pub fn recover_current_project_servers() {
             continue;
         }
 
-        // Only recover if we don't already have metadata for this server
-        if load_running_info(&proc.server_name).is_some() {
-            continue;
-        }
-
+        validate_server_name(&proc.server_name)?;
         let info = ServerInfo {
             name: proc.server_name,
             pid: proc.pid,
@@ -721,11 +916,20 @@ pub fn recover_current_project_servers() {
             engine: Engine::Clickhouse,
             container_id: None,
         };
-        let _ = save_server_info(&info);
+        recover_clickhouse_info_locked(&info, lock)?;
     }
 
     // Also recover orphaned Postgres containers belonging to this project.
-    docker::recover_project_postgres_blocking(&current_dir);
+    docker::recover_project_postgres_blocking(&current_dir, lock)
+}
+
+fn recover_clickhouse_info_locked(info: &ServerInfo, lock: &MetadataLock) -> Result<()> {
+    // A corrupt existing file is an error, not an absent entry that recovery
+    // is allowed to overwrite.
+    if load_running_info_locked(&info.name, lock)?.is_none() {
+        save_server_info_locked(info, lock)?;
+    }
+    Ok(())
 }
 
 /// A server entry for global listing — always running (discovered via process inspection).
@@ -774,6 +978,20 @@ pub fn kill_server_by_pid(pid: u32) -> Result<()> {
 mod tests {
     use super::*;
 
+    fn test_info(pid: u32, version: &str) -> ServerInfo {
+        ServerInfo {
+            name: "default".into(),
+            pid,
+            version: version.into(),
+            http_port: 8123,
+            tcp_port: 9000,
+            started_at: "1700000000".into(),
+            cwd: "/tmp/project".into(),
+            engine: Engine::Clickhouse,
+            container_id: None,
+        }
+    }
+
     #[test]
     fn engine_serializes_lowercase() {
         assert_eq!(
@@ -823,6 +1041,240 @@ mod tests {
         assert!(json.contains("\"engine\":\"postgres\""));
     }
 
+    #[test]
+    fn selected_metadata_reports_partial_json() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("default.json");
+        std::fs::write(&path, br#"{"name":"default","pid":12"#).unwrap();
+
+        let error = load_info_at(&path).unwrap_err();
+        assert!(matches!(error, Error::ServerMetadataParse { .. }));
+        assert!(error.to_string().contains("not valid JSON"));
+        assert!(error.to_string().contains("default.json"));
+    }
+
+    #[test]
+    fn selected_metadata_reports_invalid_utf8() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("default.json");
+        std::fs::write(&path, [0xff, 0xfe, 0xfd]).unwrap();
+
+        let error = load_info_at(&path).unwrap_err();
+        assert!(matches!(error, Error::ServerMetadataUtf8 { .. }));
+        assert!(error.to_string().contains("not valid UTF-8"));
+    }
+
+    #[test]
+    fn listing_ignores_json_directories() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("default.json");
+        std::fs::create_dir(&path).unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+
+        assert!(list_all_servers_locked(&lock).unwrap().is_empty());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn selected_metadata_reports_permission_denied() {
+        use std::os::unix::fs::PermissionsExt;
+
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("default.json");
+        std::fs::write(&path, b"{}").unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o000)).unwrap();
+
+        let error = load_info_at(&path).unwrap_err();
+        let Error::ServerMetadataPermission { source, .. } = error else {
+            panic!("expected metadata permission error");
+        };
+        assert_eq!(source.kind(), std::io::ErrorKind::PermissionDenied);
+    }
+
+    #[test]
+    fn metadata_lock_directory_failure_is_actionable() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock_directory = directory.path().join("servers");
+        std::fs::write(&lock_directory, b"not a directory").unwrap();
+
+        let error = match MetadataLock::acquire_at(&lock_directory) {
+            Ok(_) => panic!("metadata lock acquisition unexpectedly succeeded"),
+            Err(error) => error,
+        };
+
+        assert!(std::error::Error::source(&error).is_some());
+        assert!(matches!(
+            error,
+            Error::ServerLock {
+                operation: "create the server metadata lock directory",
+                path,
+                source,
+                ..
+            } if path == lock_directory
+                && source.kind() == std::io::ErrorKind::AlreadyExists
+        ));
+    }
+
+    #[test]
+    fn atomic_save_ignores_interrupted_sibling_temp_files() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        let stale_temp = directory.path().join(".metadata-interrupted-write");
+        std::fs::write(&stale_temp, br#"{"name":"default""#).unwrap();
+
+        save_server_info_locked(&test_info(0, "25.12.1.1"), &lock).unwrap();
+        let entries = list_all_servers_locked(&lock).unwrap();
+
+        assert_eq!(entries.len(), 1);
+        assert_eq!(entries[0].name, "default");
+        assert_eq!(entries[0].info.as_ref().unwrap().version, "25.12.1.1");
+        assert_eq!(
+            std::fs::read_to_string(stale_temp).unwrap(),
+            r#"{"name":"default""#
+        );
+    }
+
+    #[test]
+    fn directory_sync_failure_after_persist_keeps_committed_metadata() {
+        let directory = tempfile::tempdir().unwrap();
+        let info = test_info(0, "committed");
+
+        save_server_info_at_with_sync(directory.path(), &info, |_, path| {
+            Err(metadata_write_error(
+                path,
+                std::io::Error::other("injected directory sync failure"),
+            ))
+        })
+        .unwrap();
+
+        let stored = load_info_at(&directory.path().join("default.json"))
+            .unwrap()
+            .unwrap();
+        assert_eq!(stored.version, "committed");
+    }
+
+    #[test]
+    fn advisory_count_ignores_corrupt_unrelated_metadata() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        let mut healthy = test_info(std::process::id(), "25.12.1.1");
+        healthy.name = "healthy".into();
+        save_server_info_locked(&healthy, &lock).unwrap();
+        std::fs::write(directory.path().join("corrupt.json"), b"not json").unwrap();
+
+        assert_eq!(advisory_running_server_count_locked(&lock), 1);
+        assert!(matches!(
+            list_all_servers_locked(&lock),
+            Err(Error::ServerMetadataParse { .. })
+        ));
+    }
+
+    #[test]
+    fn concurrent_unlocked_readers_never_observe_partial_json() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        save_server_info_locked(&test_info(0, "initial"), &lock).unwrap();
+        drop(lock);
+        let writer_dir = directory.path().to_path_buf();
+        let metadata_path = directory.path().join("default.json");
+
+        let writer = std::thread::spawn(move || {
+            for generation in 0..200 {
+                let lock = MetadataLock::acquire_at(&writer_dir).unwrap();
+                save_server_info_locked(&test_info(0, &format!("generation-{generation}")), &lock)
+                    .unwrap();
+            }
+        });
+        for _ in 0..1_000 {
+            let bytes = std::fs::read(&metadata_path).unwrap();
+            serde_json::from_slice::<ServerInfo>(&bytes).unwrap();
+        }
+        writer.join().unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn metadata_write_permission_failure_is_not_discarded() {
+        use std::os::unix::fs::PermissionsExt;
+
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        save_server_info_locked(&test_info(0, "preserved"), &lock).unwrap();
+        std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o555)).unwrap();
+
+        let error = save_server_info_locked(&test_info(0, "blocked"), &lock).unwrap_err();
+        assert!(matches!(error, Error::ServerMetadataWrite { .. }));
+
+        std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o755)).unwrap();
+        assert_eq!(
+            load_info_locked("default", &lock).unwrap().unwrap().version,
+            "preserved"
+        );
+    }
+
+    #[test]
+    fn stale_pid_normalization_is_durable() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        save_server_info_locked(&test_info(u32::MAX, "25.12.1.1"), &lock).unwrap();
+
+        let entry = server_entry_locked("default", &lock).unwrap().unwrap();
+        let normalized = entry.info.unwrap();
+        assert!(!entry.running);
+        assert_eq!(normalized.pid, 0);
+        assert!(normalized.version.is_empty());
+        assert_eq!(load_info_locked("default", &lock).unwrap().unwrap().pid, 0);
+    }
+
+    #[test]
+    fn recovery_does_not_overwrite_corrupt_live_metadata() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        let path = directory.path().join("default.json");
+        let partial = br#"{"name":"default""#;
+        std::fs::write(&path, partial).unwrap();
+
+        let error =
+            recover_clickhouse_info_locked(&test_info(std::process::id(), "recovered"), &lock)
+                .unwrap_err();
+
+        assert!(matches!(error, Error::ServerMetadataParse { .. }));
+        assert_eq!(std::fs::read(path).unwrap(), partial);
+    }
+
+    #[test]
+    fn restart_waiting_during_normalization_commits_last() {
+        let directory = tempfile::tempdir().unwrap();
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        save_server_info_locked(&test_info(u32::MAX, "stale"), &lock).unwrap();
+        let restart_dir = directory.path().to_path_buf();
+        let mut restart = None;
+
+        let normalized = server_entry_locked_with("default", &lock, || {
+            restart = Some(std::thread::spawn(move || {
+                let restart_lock = MetadataLock::acquire_at(&restart_dir).unwrap();
+                save_server_info_locked(&test_info(std::process::id(), "restarted"), &restart_lock)
+                    .unwrap();
+            }));
+        })
+        .unwrap()
+        .unwrap();
+        assert_eq!(normalized.info.unwrap().pid, 0);
+        drop(lock);
+        restart.unwrap().join().unwrap();
+
+        let lock = MetadataLock::acquire_at(directory.path()).unwrap();
+        let final_info = load_info_locked("default", &lock).unwrap().unwrap();
+        assert_eq!(final_info.pid, std::process::id());
+        assert_eq!(final_info.version, "restarted");
+    }
+
     #[test]
     fn stopped_and_out_of_range_pids_are_never_alive() {
         assert!(!is_process_alive(0));
diff --git a/crates/clickhousectl/src/version_manager/install.rs b/crates/clickhousectl/src/version_manager/install.rs
index 3ac0cfe0..22693808 100644
--- a/crates/clickhousectl/src/version_manager/install.rs
+++ b/crates/clickhousectl/src/version_manager/install.rs
@@ -138,7 +138,7 @@ pub async fn install_resolved(
     };
 
     let commit_lock = CommitLock::acquire(&versions_dir).await?;
-    let replaced_existing = commit_staged_install_locked(
+    let (_, notify_running_servers) = commit_staged_install_locked(
         &commit_lock,
         &versions_dir,
         &staging,
@@ -147,12 +147,14 @@ pub async fn install_resolved(
         is_master,
         platform,
         master_head.as_ref(),
+        version_in_use_by_running_server,
         |_| Ok(()),
     )?;
 
     // Replacing a build on disk never affects already-running servers (they keep
     // executing the old binary) — just say so, so the swap isn't silent.
-    if is_master && replaced_existing && version_in_use_by_running_server(&exact_version) {
+    drop(commit_lock);
+    if notify_running_servers {
         eprintln!(
             "Note: running servers keep using the previous {} build until restarted",
             exact_version
@@ -184,8 +186,9 @@ fn commit_staged_install_locked(
     is_master: bool,
     platform: &Platform,
     master_head: Option<&master::HeadInfo>,
+    version_in_use: impl FnOnce(&str) -> Result<bool>,
     mut checkpoint: impl FnMut(CommitCheckpoint) -> Result<()>,
-) -> Result<bool> {
+) -> Result<(bool, bool)> {
     let version_dir = versions_dir.join(exact_version);
     let target_binary = version_dir.join("clickhouse");
     let target_metadata = match std::fs::symlink_metadata(&version_dir) {
@@ -206,6 +209,7 @@ fn commit_staged_install_locked(
     if replaced_existing && !force && !is_master {
         return Err(Error::VersionAlreadyInstalled(exact_version.to_string()));
     }
+    let notify_running_servers = is_master && replaced_existing && version_in_use(exact_version)?;
 
     File::open(staging.binary_path())?.sync_all()?;
     sync_directory(staging.payload())?;
@@ -235,7 +239,7 @@ fn commit_staged_install_locked(
         );
     }
 
-    Ok(replaced_existing)
+    Ok((replaced_existing, notify_running_servers))
 }
 
 /// Like `install_resolved`, but returns the existing version instead of erroring
@@ -280,11 +284,11 @@ fn is_installed(binary_path: &Path) -> bool {
 /// Whether a running managed server (in the current project) was started from
 /// this version. Recovers orphans first (like `local remove`) so a server that
 /// lost its metadata file is still counted.
-fn version_in_use_by_running_server(version: &str) -> bool {
-    crate::local::server::recover_current_project_servers();
-    crate::local::server::list_running_servers()
+fn version_in_use_by_running_server(version: &str) -> Result<bool> {
+    crate::local::server::recover_current_project_servers()?;
+    Ok(crate::local::server::list_running_servers()?
         .iter()
-        .any(|s| s.version == version)
+        .any(|s| s.version == version))
 }
 
 /// Detect the version of a clickhouse binary by running `./clickhouse --version`
@@ -632,11 +636,12 @@ mod tests {
             true,
             &test_platform(),
             Some(&head),
+            |_| Ok(false),
             |_| Ok(()),
         )
         .unwrap();
 
-        assert!(!replaced);
+        assert!(!replaced.0);
         assert_eq!(
             fs::read(versions_dir.join("26.5.1.1/clickhouse")).unwrap(),
             b"complete-master-build"
@@ -644,6 +649,48 @@ mod tests {
         assert!(!versions_dir.join(".master-builds.json").exists());
     }
 
+    #[test]
+    fn metadata_failure_does_not_replace_existing_master_binary() {
+        let temp = tempfile::tempdir().unwrap();
+        let versions_dir = temp.path().join("versions");
+        let version_dir = versions_dir.join("26.5.1.1");
+        fs::create_dir_all(&version_dir).unwrap();
+        let installed_binary = version_dir.join("clickhouse");
+        fs::write(&installed_binary, b"existing master build").unwrap();
+        let staging = InstallStaging::create(&versions_dir).unwrap();
+        fs::write(staging.binary_path(), b"replacement master build").unwrap();
+        let lock = CommitLock::acquire_blocking(&versions_dir).unwrap();
+
+        let error = commit_staged_install_locked(
+            &lock,
+            &versions_dir,
+            &staging,
+            "26.5.1.1",
+            true,
+            true,
+            &test_platform(),
+            None,
+            |_| {
+                Err(Error::ServerMetadataRead {
+                    path: PathBuf::from("broken.json"),
+                    source: std::io::Error::other("metadata unavailable"),
+                })
+            },
+            |_| Ok(()),
+        )
+        .unwrap_err();
+
+        assert!(matches!(error, Error::ServerMetadataRead { .. }));
+        assert_eq!(
+            fs::read(installed_binary).unwrap(),
+            b"existing master build"
+        );
+        assert_eq!(
+            fs::read(staging.binary_path()).unwrap(),
+            b"replacement master build"
+        );
+    }
+
     #[test]
     #[ignore = "subprocess helper for atomic install tests"]
     fn atomic_install_process_helper() {
@@ -682,6 +729,7 @@ mod tests {
             true,
             &test_platform(),
             Some(&head),
+            |_| Ok(false),
             |checkpoint| {
                 let checkpoint_name = match checkpoint {
                     CommitCheckpoint::SidecarInvalidated => "invalidated",
diff --git a/crates/clickhousectl/tests/local_postgres_readiness_test.rs b/crates/clickhousectl/tests/local_postgres_readiness_test.rs
index 88e25e17..c2187c6f 100644
--- a/crates/clickhousectl/tests/local_postgres_readiness_test.rs
+++ b/crates/clickhousectl/tests/local_postgres_readiness_test.rs
@@ -1,6 +1,7 @@
 //! Subprocess coverage for local Postgres readiness through a fake Docker API.
 
 use std::collections::{HashMap, VecDeque};
+use std::fs::OpenOptions;
 use std::io::{ErrorKind, Read, Write};
 use std::net::Shutdown;
 use std::os::unix::net::{UnixListener, UnixStream};
@@ -32,6 +33,7 @@ struct DockerScenario {
     readiness_create_errors: usize,
     logs: Vec<String>,
     write_partial_data: bool,
+    create_metadata_directory_on_start: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -39,6 +41,7 @@ struct DockerRequest {
     method: String,
     path: String,
     body: String,
+    metadata_lock_available: bool,
 }
 
 struct FakeDocker {
@@ -70,6 +73,7 @@ impl FakeDocker {
                 readiness_create_errors,
                 logs,
                 write_partial_data,
+                create_metadata_directory_on_start,
             } = scenario;
             let mut started = false;
             let mut next_exec = 0_usize;
@@ -94,7 +98,8 @@ impl FakeDocker {
                 stream
                     .set_read_timeout(Some(Duration::from_secs(1)))
                     .expect("set fake Docker read timeout");
-                let request = read_request(&mut stream);
+                let mut request = read_request(&mut stream);
+                request.metadata_lock_available = metadata_lock_available(&project);
                 thread_requests.lock().unwrap().push(request.clone());
 
                 match (request.method.as_str(), request.path.as_str()) {
@@ -102,13 +107,19 @@ impl FakeDocker {
                     ("GET", path) if path.starts_with("/containers/json?") => {
                         write_json(&mut stream, 200, "[]")
                     }
-                    ("GET", "/images/postgres:18/json") => write_json(&mut stream, 200, "{}"),
+                    ("GET", "/images/postgres:18/json") => {
+                        inject_metadata_during_image_inspect(&project);
+                        write_json(&mut stream, 200, "{}");
+                    }
                     ("GET", "/images/alpine:latest/json") => write_json(&mut stream, 200, "{}"),
                     ("GET", path)
                         if path.starts_with("/containers/clickhousectl-pg-default-18/json") =>
                     {
                         write_json(&mut stream, 404, r#"{"message":"No such container"}"#)
                     }
+                    ("GET", path) if path.starts_with("/containers/concurrent-id/json") => {
+                        write_json(&mut stream, 404, r#"{"message":"No such container"}"#)
+                    }
                     ("POST", path) if path.starts_with("/containers/create?") => {
                         let body: serde_json::Value = serde_json::from_str(&request.body)
                             .expect("container create body JSON");
@@ -138,6 +149,12 @@ impl FakeDocker {
                                 std::fs::write(&partial_data_path, "partial PGDATA")
                                     .expect("write partial PGDATA marker");
                             }
+                            if create_metadata_directory_on_start {
+                                let instance_dir =
+                                    partial_data_path.parent().unwrap().parent().unwrap();
+                                std::fs::create_dir(instance_dir.with_extension("json"))
+                                    .expect("create metadata failure directory");
+                            }
                             write_response(&mut stream, 204, "application/json", b"");
                         } else {
                             write_json(
@@ -161,6 +178,13 @@ impl FakeDocker {
                         );
                         write_json(&mut stream, 200, &body);
                     }
+                    ("GET", path) if path.starts_with("/containers/dotenv-id/json") => {
+                        write_json(
+                            &mut stream,
+                            200,
+                            r#"{"Id":"dotenv-id","State":{"Running":true},"Config":{"Env":["POSTGRES_USER=dotenv-user","POSTGRES_PASSWORD=dotenv-secret","POSTGRES_DB=dotenv-database"]}}"#,
+                        );
+                    }
                     ("POST", "/containers/pg-id/exec") => {
                         if readiness_create_errors > 0 {
                             readiness_create_errors -= 1;
@@ -283,9 +307,59 @@ fn read_request(stream: &mut UnixStream) -> DockerRequest {
         method: request_parts.next().expect("HTTP method").to_string(),
         path: request_parts.next().expect("HTTP path").to_string(),
         body: String::from_utf8_lossy(&bytes[header_end..header_end + content_length]).into_owned(),
+        metadata_lock_available: false,
     }
 }
 
+fn metadata_lock_available(project: &Path) -> bool {
+    let path = project.join(".clickhouse/servers/.metadata.lock");
+    let Ok(file) = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(path)
+    else {
+        return false;
+    };
+    file.try_lock().is_ok()
+}
+
+fn inject_metadata_during_image_inspect(project: &Path) {
+    let marker = project.join("inject-metadata-during-image-inspect");
+    if !marker.exists() {
+        return;
+    }
+
+    let servers = project.join(".clickhouse/servers");
+    let lock = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(servers.join(".metadata.lock"))
+        .expect("open metadata lock for concurrent write");
+    lock.lock()
+        .expect("acquire metadata lock for concurrent write");
+    let metadata = serde_json::json!({
+        "name": "default-pg18",
+        "pid": 0,
+        "version": "postgres:18",
+        "http_port": 0,
+        "tcp_port": 5432,
+        "started_at": "concurrent",
+        "cwd": project.canonicalize().unwrap(),
+        "engine": "postgres",
+        "container_id": "concurrent-id"
+    });
+    std::fs::write(
+        servers.join("default-pg18.json"),
+        serde_json::to_vec_pretty(&metadata).unwrap(),
+    )
+    .expect("write concurrent metadata");
+    std::fs::remove_file(marker).expect("remove metadata injection marker");
+}
+
 fn write_json(stream: &mut UnixStream, status: u16, body: &str) {
     write_response(stream, status, "application/json", body.as_bytes());
 }
@@ -351,6 +425,28 @@ fn write_resumed_server(project: &Path) {
     .expect("write resumed server metadata");
 }
 
+fn write_dotenv_server(project: &Path) {
+    let servers = project.join(".clickhouse/servers");
+    std::fs::create_dir_all(servers.join("default-pg18/data"))
+        .expect("create dotenv server data directory");
+    let metadata = serde_json::json!({
+        "name": "default-pg18",
+        "pid": 0,
+        "version": "postgres:18",
+        "http_port": 0,
+        "tcp_port": 5432,
+        "started_at": "before-dotenv",
+        "cwd": project.canonicalize().unwrap(),
+        "engine": "postgres",
+        "container_id": "dotenv-id"
+    });
+    std::fs::write(
+        servers.join("default-pg18.json"),
+        serde_json::to_vec_pretty(&metadata).unwrap(),
+    )
+    .expect("write dotenv server metadata");
+}
+
 fn run_start(
     scenario: DockerScenario,
     resumed: bool,
@@ -454,6 +550,7 @@ fn fresh_start_waits_for_delayed_postgres_readiness_without_exposing_password()
             readiness_create_errors: 1,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         false,
         false,
@@ -471,6 +568,10 @@ fn fresh_start_waits_for_delayed_postgres_readiness_without_exposing_password()
     let probes = readiness_requests(&requests);
     assert_eq!(probes.len(), 3);
     for probe in probes {
+        assert!(
+            probe.metadata_lock_available,
+            "metadata lock was held during readiness request: {probe:?}"
+        );
         let body: serde_json::Value = serde_json::from_str(&probe.body).expect("exec body JSON");
         assert_eq!(
             body["Cmd"],
@@ -488,6 +589,130 @@ fn fresh_start_waits_for_delayed_postgres_readiness_without_exposing_password()
         assert!(!probe.body.contains("fresh-secret"));
         assert!(body["Env"].is_null());
     }
+    let image_inspect = requests
+        .iter()
+        .find(|request| request.path == "/images/postgres:18/json")
+        .expect("fresh image inspection request");
+    assert!(
+        image_inspect.metadata_lock_available,
+        "metadata lock was held during image inspection: {image_inspect:?}"
+    );
+}
+
+#[test]
+fn postgres_dotenv_releases_metadata_lock_before_docker_credentials_read() {
+    let home = tempfile::tempdir().expect("create home tempdir");
+    let project = tempfile::tempdir().expect("create project tempdir");
+    write_dotenv_server(project.path());
+    let socket_path = home.path().join("docker.sock");
+    let docker = FakeDocker::start(
+        &socket_path,
+        project.path(),
+        DockerScenario {
+            existing: true,
+            outcome: ContainerOutcome::Running,
+            start_statuses: vec![],
+            remove_statuses: vec![],
+            readiness_exit_codes: vec![],
+            readiness_create_errors: 0,
+            logs: vec![],
+            write_partial_data: false,
+            create_metadata_directory_on_start: false,
+        },
+    );
+
+    let output = Command::new(clickhousectl_binary())
+        .env_clear()
+        .env("DO_NOT_TRACK", "1")
+        .env("HOME", home.path())
+        .env("DOCKER_HOST", format!("unix://{}", socket_path.display()))
+        .current_dir(project.path())
+        .args(["local", "--json", "postgres", "dotenv", "--name", "default"])
+        .output()
+        .expect("run postgres dotenv");
+    let requests = docker.requests();
+
+    assert!(
+        output.status.success(),
+        "stderr: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let inspections: Vec<_> = requests
+        .iter()
+        .filter(|request| request.path == "/containers/dotenv-id/json")
+        .collect();
+    assert!(inspections.len() >= 2, "requests: {requests:?}");
+    assert!(
+        inspections.last().unwrap().metadata_lock_available,
+        "metadata lock was held during credential read: {inspections:?}"
+    );
+}
+
+#[test]
+fn postgres_start_revalidates_metadata_after_image_inspection() {
+    let home = tempfile::tempdir().expect("create home tempdir");
+    let project = tempfile::tempdir().expect("create project tempdir");
+    std::fs::write(
+        project.path().join("inject-metadata-during-image-inspect"),
+        b"inject",
+    )
+    .expect("write metadata injection marker");
+    let socket_path = home.path().join("docker.sock");
+    let docker = FakeDocker::start(
+        &socket_path,
+        project.path(),
+        DockerScenario {
+            existing: false,
+            outcome: ContainerOutcome::Running,
+            start_statuses: vec![],
+            remove_statuses: vec![],
+            readiness_exit_codes: vec![],
+            readiness_create_errors: 0,
+            logs: vec![],
+            write_partial_data: false,
+            create_metadata_directory_on_start: false,
+        },
+    );
+    let port = reserve_port().to_string();
+
+    let output = Command::new(clickhousectl_binary())
+        .env_clear()
+        .env("DO_NOT_TRACK", "1")
+        .env("HOME", home.path())
+        .env("DOCKER_HOST", format!("unix://{}", socket_path.display()))
+        .current_dir(project.path())
+        .args([
+            "local",
+            "postgres",
+            "start",
+            "--name",
+            "default",
+            "--version",
+            "18",
+            "--port",
+            &port,
+        ])
+        .output()
+        .expect("run postgres start");
+    let requests = docker.requests();
+
+    assert_eq!(output.status.code(), Some(1));
+    assert!(
+        String::from_utf8_lossy(&output.stderr).contains("container is gone"),
+        "stderr: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert!(
+        !requests
+            .iter()
+            .any(|request| request.path.starts_with("/containers/create")),
+        "start ignored concurrently committed metadata: {requests:?}"
+    );
+    let metadata: serde_json::Value = serde_json::from_slice(
+        &std::fs::read(project.path().join(".clickhouse/servers/default-pg18.json")).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(metadata["container_id"], "concurrent-id");
 }
 
 #[test]
@@ -502,6 +727,7 @@ fn resumed_start_also_waits_for_postgres_readiness() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         true,
         false,
@@ -538,6 +764,7 @@ fn wall_clock_timeout_fails_and_rolls_back_fresh_data() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         false,
         false,
@@ -585,6 +812,7 @@ fn immediate_exit_reports_bounded_logs_and_error_telemetry_without_setup_success
             readiness_create_errors: 0,
             logs,
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         false,
         true,
@@ -644,6 +872,7 @@ fn failed_fresh_start_preserves_preexisting_data_and_recovery_metadata() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         false,
         false,
@@ -693,6 +922,7 @@ fn incomplete_container_cleanup_retains_pgdata_and_recovery_metadata() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
         false,
         false,
@@ -759,6 +989,7 @@ fn create_success_start_failure_rolls_back_exact_container_and_fresh_data() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
     );
 
@@ -792,6 +1023,7 @@ fn initialization_timeout_removes_partial_pgdata() {
             readiness_create_errors: 0,
             logs: vec!["database system is starting up".to_string()],
             write_partial_data: true,
+            create_metadata_directory_on_start: false,
         },
     );
 
@@ -809,15 +1041,8 @@ fn initialization_timeout_removes_partial_pgdata() {
 
 #[test]
 fn metadata_failure_uses_the_fresh_start_rollback() {
-    use std::os::unix::fs::symlink;
-
     let home = tempfile::tempdir().expect("create home tempdir");
     let project = tempfile::tempdir().expect("create project tempdir");
-    let servers = project.path().join(".clickhouse/servers");
-    let metadata_target = project.path().join("metadata-target-directory");
-    std::fs::create_dir_all(&servers).expect("create servers directory");
-    std::fs::create_dir(&metadata_target).expect("create metadata failure target");
-    symlink(&metadata_target, metadata_path(project.path())).expect("create metadata symlink");
 
     let socket_path = home.path().join("docker.sock");
     let docker = FakeDocker::start(
@@ -832,6 +1057,7 @@ fn metadata_failure_uses_the_fresh_start_rollback() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: true,
+            create_metadata_directory_on_start: true,
         },
     );
 
@@ -839,10 +1065,13 @@ fn metadata_failure_uses_the_fresh_start_rollback() {
     let requests = docker.requests();
 
     assert_eq!(output.status.code(), Some(1));
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(stderr.contains("Failed to durably update server metadata"));
+    assert!(stderr.contains("failed to remove metadata"));
     request_index(&requests, "DELETE", "/containers/pg-id?");
     assert!(readiness_requests(&requests).is_empty());
     assert!(!fresh_instance_dir(project.path()).exists());
-    assert!(!metadata_path(project.path()).exists());
+    assert!(metadata_path(project.path()).is_dir());
 }
 
 #[test]
@@ -862,6 +1091,7 @@ fn retry_after_rolled_back_start_failure_succeeds_cleanly() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
     );
 
@@ -912,6 +1142,7 @@ fn resume_failure_preserves_existing_container_metadata_and_data() {
             readiness_create_errors: 0,
             logs: vec!["resume failed".to_string()],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
     );
 
@@ -944,6 +1175,7 @@ fn cleanup_failure_keeps_primary_start_error_and_adds_diagnostics() {
             readiness_create_errors: 0,
             logs: vec![],
             write_partial_data: false,
+            create_metadata_directory_on_start: false,
         },
     );
 
@@ -952,7 +1184,9 @@ fn cleanup_failure_keeps_primary_start_error_and_adds_diagnostics() {
 
     assert_eq!(output.status.code(), Some(1));
     let stderr = String::from_utf8_lossy(&output.stderr);
-    let primary = stderr.find("start failed by test").expect("primary error");
+    let primary = stderr
+        .find("start failed by test")
+        .unwrap_or_else(|| panic!("primary error missing: {stderr}"));
     let rollback = stderr
         .find("Postgres startup rollback incomplete")
         .expect("rollback diagnostics");
diff --git a/crates/clickhousectl/tests/local_server_metadata_test.rs b/crates/clickhousectl/tests/local_server_metadata_test.rs
new file mode 100644
index 00000000..66e7c841
--- /dev/null
+++ b/crates/clickhousectl/tests/local_server_metadata_test.rs
@@ -0,0 +1,351 @@
+//! Regression coverage for atomic, concurrency-safe server metadata (issue #472).
+
+use serde_json::{Value, json};
+use std::fs::OpenOptions;
+use std::os::unix::fs::PermissionsExt;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output, Stdio};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+const VERSION: &str = "25.12.9.61";
+
+fn clickhousectl_binary() -> PathBuf {
+    PathBuf::from(env!("CARGO_BIN_EXE_clickhousectl"))
+}
+
+fn command(project: &Path, home: &Path) -> Command {
+    let mut command = Command::new(clickhousectl_binary());
+    command
+        .env("DO_NOT_TRACK", "1")
+        .env("HOME", home)
+        .env(
+            "FAKE_CLICKHOUSE_PID_FILE",
+            project.join("fake-clickhouse-pids"),
+        )
+        .current_dir(project);
+    command
+}
+
+fn run(project: &Path, home: &Path, args: &[&str]) -> Output {
+    command(project, home)
+        .args(args)
+        .output()
+        .expect("run clickhousectl")
+}
+
+fn servers_dir(project: &Path) -> PathBuf {
+    project.join(".clickhouse/servers")
+}
+
+fn write_metadata(project: &Path, bytes: &[u8]) -> PathBuf {
+    let directory = servers_dir(project);
+    std::fs::create_dir_all(directory.join("default/data")).expect("create server data dir");
+    let path = directory.join("default.json");
+    std::fs::write(&path, bytes).expect("write metadata");
+    path
+}
+
+fn valid_metadata(project: &Path, pid: u32) -> Vec<u8> {
+    serde_json::to_vec_pretty(&json!({
+        "name": "default",
+        "pid": pid,
+        "version": VERSION,
+        "http_port": 8123,
+        "tcp_port": 9000,
+        "started_at": "1700000000",
+        "cwd": project.display().to_string(),
+        "engine": "clickhouse"
+    }))
+    .unwrap()
+}
+
+fn install_fake_clickhouse(home: &Path) {
+    let binary = home.join(format!(".clickhouse/versions/{VERSION}/clickhouse"));
+    std::fs::create_dir_all(binary.parent().unwrap()).expect("create fake version dir");
+    std::fs::write(
+        &binary,
+        b"#!/bin/sh\ncase \"$1\" in\nserver) printf '%s\\n' \"$$\" >> \"$FAKE_CLICKHOUSE_PID_FILE\"; trap 'exit 0' TERM INT; while :; do sleep 0.1; done ;;\nclient) exit 0 ;;\nesac\n",
+    )
+    .expect("write fake ClickHouse");
+    std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o755))
+        .expect("make fake ClickHouse executable");
+}
+
+struct ProcessFileGuard(PathBuf);
+
+impl Drop for ProcessFileGuard {
+    fn drop(&mut self) {
+        let Ok(contents) = std::fs::read_to_string(&self.0) else {
+            return;
+        };
+        for pid in contents.lines().filter_map(|line| line.parse::<i32>().ok()) {
+            unsafe {
+                libc::kill(pid, libc::SIGKILL);
+            }
+        }
+    }
+}
+
+#[test]
+fn selected_partial_json_is_not_reported_as_not_running() {
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    write_metadata(project.path(), br#"{"name":"default","pid":123"#);
+
+    let output = run(
+        project.path(),
+        home.path(),
+        &["local", "client", "--name", "default"],
+    );
+
+    assert_eq!(output.status.code(), Some(1));
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(stderr.contains("not valid JSON"), "stderr: {stderr}");
+    assert!(stderr.contains("default.json"), "stderr: {stderr}");
+    assert!(!stderr.contains("is not running"), "stderr: {stderr}");
+}
+
+#[test]
+fn selected_invalid_utf8_is_actionable() {
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    write_metadata(project.path(), &[0xff, 0xfe, 0xfd]);
+
+    let output = run(
+        project.path(),
+        home.path(),
+        &["local", "client", "--name", "default"],
+    );
+
+    assert_eq!(output.status.code(), Some(1));
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(stderr.contains("not valid UTF-8"), "stderr: {stderr}");
+    assert!(!stderr.contains("is not running"), "stderr: {stderr}");
+}
+
+#[cfg(unix)]
+#[test]
+fn selected_permission_failure_is_actionable() {
+    if unsafe { libc::geteuid() } == 0 {
+        return;
+    }
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    let metadata = write_metadata(project.path(), &valid_metadata(project.path(), 0));
+    std::fs::set_permissions(&metadata, std::fs::Permissions::from_mode(0o000))
+        .expect("remove metadata permissions");
+
+    let output = run(
+        project.path(),
+        home.path(),
+        &["local", "client", "--name", "default"],
+    );
+
+    std::fs::set_permissions(&metadata, std::fs::Permissions::from_mode(0o600))
+        .expect("restore metadata permissions");
+    assert_eq!(output.status.code(), Some(1));
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(
+        stderr.contains("Permission denied reading server metadata"),
+        "stderr: {stderr}"
+    );
+    assert!(stderr.contains("Check ownership and file permissions"));
+    assert!(!stderr.contains("is not running"), "stderr: {stderr}");
+}
+
+#[test]
+fn list_ignores_stale_temp_but_rejects_corrupt_live_entry() {
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    let directory = servers_dir(project.path());
+    std::fs::create_dir_all(&directory).expect("create servers dir");
+    std::fs::write(
+        directory.join(".metadata-interrupted-write"),
+        br#"{"name":"default""#,
+    )
+    .expect("write stale temp");
+
+    let absent = run(
+        project.path(),
+        home.path(),
+        &["local", "--json", "server", "list"],
+    );
+    assert!(
+        absent.status.success(),
+        "stderr: {}",
+        String::from_utf8_lossy(&absent.stderr)
+    );
+    let body: Value = serde_json::from_slice(&absent.stdout).expect("parse list JSON");
+    assert_eq!(body["total_servers"], 0);
+
+    std::fs::write(directory.join("default.json"), b"{").expect("write corrupt live entry");
+    let corrupt = run(
+        project.path(),
+        home.path(),
+        &["local", "--json", "server", "list"],
+    );
+    assert_eq!(corrupt.status.code(), Some(1));
+    assert!(String::from_utf8_lossy(&corrupt.stderr).contains("not valid JSON"));
+}
+
+#[test]
+fn corrupt_unrelated_metadata_does_not_block_start() {
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    install_fake_clickhouse(home.path());
+    let _processes = ProcessFileGuard(project.path().join("fake-clickhouse-pids"));
+    let metadata = write_metadata(project.path(), b"not json");
+
+    let output = run(
+        project.path(),
+        home.path(),
+        &[
+            "local",
+            "--json",
+            "server",
+            "start",
+            "--name",
+            "unrelated",
+            "--version",
+            VERSION,
+            "--no-wait",
+        ],
+    );
+
+    assert!(
+        output.status.success(),
+        "stderr: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert_eq!(std::fs::read(metadata).unwrap(), b"not json");
+}
+
+#[test]
+fn concurrent_start_client_and_stop_leave_valid_metadata() {
+    let project = tempfile::tempdir().expect("create project tempdir");
+    let home = tempfile::tempdir().expect("create home tempdir");
+    install_fake_clickhouse(home.path());
+    let _processes = ProcessFileGuard(project.path().join("fake-clickhouse-pids"));
+
+    let initial = run(
+        project.path(),
+        home.path(),
+        &[
+            "local",
+            "--json",
+            "server",
+            "start",
+            "--name",
+            "default",
+            "--version",
+            VERSION,
+            "--no-wait",
+        ],
+    );
+    assert!(
+        initial.status.success(),
+        "stderr: {}",
+        String::from_utf8_lossy(&initial.stderr)
+    );
+    let initial_body: Value = serde_json::from_slice(&initial.stdout).expect("parse start JSON");
+    let initial_pid = initial_body["pid"].as_u64().expect("initial PID") as u32;
+
+    let directory = servers_dir(project.path());
+    let lock = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(directory.join(".metadata.lock"))
+        .expect("open metadata lock");
+    lock.lock().expect("hold metadata lock");
+
+    let mut restart = command(project.path(), home.path());
+    restart
+        .args([
+            "local",
+            "server",
+            "start",
+            "--name",
+            "default",
+            "--version",
+            VERSION,
+            "--no-wait",
+        ])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+    let restart = restart.spawn().expect("spawn concurrent restart");
+
+    let mut client = command(project.path(), home.path());
+    client
+        .args([
+            "local", "client", "--name", "default", "--query", "SELECT 1",
+        ])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+    let client = client.spawn().expect("spawn concurrent client");
+
+    let mut stop = command(project.path(), home.path());
+    stop.args(["local", "server", "stop", "default"])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+    let stop = stop.spawn().expect("spawn concurrent stop");
+
+    let keep_reading = Arc::new(AtomicBool::new(true));
+    let reader_flag = Arc::clone(&keep_reading);
+    let metadata = directory.join("default.json");
+    let reader = std::thread::spawn(move || {
+        while reader_flag.load(Ordering::Acquire) {
+            let bytes = std::fs::read(&metadata).expect("read live metadata");
+            serde_json::from_slice::<Value>(&bytes).expect("live metadata must stay complete");
+            std::thread::yield_now();
+        }
+    });
+
+    drop(lock);
+    let restart = restart.wait_with_output().expect("wait for restart");
+    let client = client.wait_with_output().expect("wait for client");
+    let stop = stop.wait_with_output().expect("wait for stop");
+    keep_reading.store(false, Ordering::Release);
+    reader.join().expect("join metadata reader");
+
+    assert!(
+        restart.status.success()
+            || String::from_utf8_lossy(&restart.stderr).contains("already running")
+            || String::from_utf8_lossy(&restart.stderr).contains("exited immediately"),
+        "restart stderr: {}",
+        String::from_utf8_lossy(&restart.stderr)
+    );
+    assert!(
+        client.status.success()
+            || String::from_utf8_lossy(&client.stderr).contains("is not running"),
+        "client stderr: {}",
+        String::from_utf8_lossy(&client.stderr)
+    );
+    assert!(
+        stop.status.success(),
+        "stop stderr: {}",
+        String::from_utf8_lossy(&stop.stderr)
+    );
+    for output in [&restart, &client, &stop] {
+        assert!(
+            !String::from_utf8_lossy(&output.stderr).contains("metadata"),
+            "unexpected metadata error: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
+    let final_metadata: Value = serde_json::from_slice(
+        &std::fs::read(directory.join("default.json")).expect("read final metadata"),
+    )
+    .expect("parse final metadata");
+    let final_pid = final_metadata["pid"].as_u64().unwrap_or(0) as u32;
+    for pid in [initial_pid, final_pid] {
+        if pid != 0 {
+            unsafe {
+                libc::kill(pid as i32, libc::SIGKILL);
+            }
+        }
+    }
+}