diff --git a/.github/workflows/pr-regression-gates.yml b/.github/workflows/pr-regression-gates.yml
index 03e0d578b..15753cd32 100644
--- a/.github/workflows/pr-regression-gates.yml
+++ b/.github/workflows/pr-regression-gates.yml
@@ -1807,6 +1807,29 @@ jobs:
             exit 1
           }
 
+      # #2994 (E2992): the settings-owner tests in the lib crate. Windows already
+      # runs them through --lib --bins --tests; this leg lists the filter, refuses
+      # an empty list, then runs exactly that list serially.
+      - name: "cargo test (E2992 settings owner, focused)"
+        working-directory: src-tauri
+        shell: bash
+        timeout-minutes: 20
+        run: |
+          set -euo pipefail
+          FILTER=e2992_
+          cargo test --locked --lib "$FILTER" -- --list 2>&1 | tee test-e2992-list.log
+          COUNT="$(grep -cE ': test$' test-e2992-list.log || true)"
+          if [ "$COUNT" -eq 0 ]; then
+            echo "::error::the ${FILTER} filter lists no tests."
+            exit 1
+          fi
+          echo "listed ${COUNT} ${FILTER} tests"
+          cargo test --locked --lib "$FILTER" -- --test-threads=1 2>&1 | tee test-e2992.log
+          grep -qE "^test result: ok\. ${COUNT} passed; 0 failed" test-e2992.log || {
+            echo "::error::expected exactly ${COUNT} passing ${FILTER} tests."
+            exit 1
+          }
+
   rust-linux-release-parity:
     name: rust-linux-release-parity
     # #1842: the Linux PR gate runs ubuntu-latest (24.04) while the Linux
@@ -2775,6 +2798,29 @@ jobs:
             exit 1
           }
 
+      # #2994 (E2992): the settings-owner tests in the lib crate. Windows already
+      # runs them through --lib --bins --tests; this leg lists the filter, refuses
+      # an empty list, then runs exactly that list serially.
+      - name: "cargo test (E2992 settings owner, focused)"
+        working-directory: src-tauri
+        shell: bash
+        timeout-minutes: 20
+        run: |
+          set -euo pipefail
+          FILTER=e2992_
+          cargo test --locked --lib "$FILTER" -- --list 2>&1 | tee test-e2992-list.log
+          COUNT="$(grep -cE ': test$' test-e2992-list.log || true)"
+          if [ "$COUNT" -eq 0 ]; then
+            echo "::error::the ${FILTER} filter lists no tests."
+            exit 1
+          fi
+          echo "listed ${COUNT} ${FILTER} tests"
+          cargo test --locked --lib "$FILTER" -- --test-threads=1 2>&1 | tee test-e2992.log
+          grep -qE "^test result: ok\. ${COUNT} passed; 0 failed" test-e2992.log || {
+            echo "::error::expected exactly ${COUNT} passing ${FILTER} tests."
+            exit 1
+          }
+
   rust-fmt:
     name: rust-fmt
     # Skip branch-deletion push events while still requiring pull request validation.
diff --git a/src-tauri/src/config/instance_artifacts.rs b/src-tauri/src/config/instance_artifacts.rs
index 289028eea..7f57b5a85 100644
--- a/src-tauri/src/config/instance_artifacts.rs
+++ b/src-tauri/src/config/instance_artifacts.rs
@@ -208,6 +208,13 @@ pub(crate) const SETTINGS_LOCK_FILE_NAME: &str = SETTINGS_LOCK_TARGET_NAME;
 /// leftover the naming migration deliberately never deletes stays ignored; the
 /// one row deliberately not composed by `layered_name!`.
 pub(crate) const SETTINGS_RETIRED_LOCK_NAME: &str = "settings.json.lock";
+/// #2994 - the settings owner's lifecycle lock: a stable sidecar, separate from
+/// the transactional `SETTINGS_LOCK_FILE_NAME`, retained for the owner's whole
+/// lifetime and never unlinked on close.
+pub(crate) const SETTINGS_OWNER_LOCK_FILE_NAME: &str = "settings.owner.no-git.lock";
+/// #2994 - the settings owner's endpoint metadata. Reserved here so the runtime
+/// artifact is ignored before any writer publishes it.
+pub(crate) const SETTINGS_OWNER_ENDPOINT_FILE_NAME: &str = "settings-owner.endpoint.no-git.json";
 /// Covers every settings migration backup instance. The concrete names are
 /// composed by their own migrations, so this glob is registry-owned and no
 /// writer imports it.
@@ -941,6 +948,12 @@ pub(crate) const INSTANCE_ARTIFACTS: &[InstanceArtifact] = &[
         disposition: Disposition::Ignore,
         comment: "# AgentsCommander: persisted session state",
     },
+    InstanceArtifact {
+        name: SETTINGS_OWNER_ENDPOINT_FILE_NAME,
+        kind: ArtifactKind::File,
+        disposition: Disposition::Ignore,
+        comment: "# AgentsCommander: settings owner endpoint metadata; machine-local runtime state",
+    },
     InstanceArtifact {
         name: SETTINGS_BACKUP_ROTATION_GLOB,
         kind: ArtifactKind::Glob,
@@ -983,6 +996,12 @@ pub(crate) const INSTANCE_ARTIFACTS: &[InstanceArtifact] = &[
         disposition: Disposition::Ignore,
         comment: "# AgentsCommander: retired pre-migration settings write lock; the naming migration never deletes it, so the leftover stays ignored",
     },
+    InstanceArtifact {
+        name: SETTINGS_OWNER_LOCK_FILE_NAME,
+        kind: ArtifactKind::File,
+        disposition: Disposition::Ignore,
+        comment: "# AgentsCommander: settings owner lifecycle lock; stable sidecar, never unlinked",
+    },
     InstanceArtifact {
         name: TELEGRAM_BRIDGE_LOG_FILE_NAME,
         kind: ArtifactKind::File,
@@ -1604,4 +1623,40 @@ mod tests {
         );
         assert_eq!(LOOP_STATE_TARGET_NAME, "loop.state.no-git.json");
     }
+
+    /// #2994 (P41) - the owner's lifecycle sidecar and endpoint metadata are Ignore
+    /// rows in sorted position, distinct from the transactional lock, and leave
+    /// every existing row in place.
+    #[test]
+    fn e2992_p41_owner_ignore() {
+        ignore_rows_are_unique_and_byte_sorted_by_name();
+        assert_eq!(SETTINGS_OWNER_LOCK_FILE_NAME, "settings.owner.no-git.lock");
+        assert_eq!(
+            SETTINGS_OWNER_ENDPOINT_FILE_NAME,
+            "settings-owner.endpoint.no-git.json"
+        );
+        assert_ne!(SETTINGS_OWNER_LOCK_FILE_NAME, SETTINGS_LOCK_FILE_NAME);
+        assert_ne!(SETTINGS_OWNER_LOCK_FILE_NAME, SETTINGS_RETIRED_LOCK_NAME);
+        for name in [
+            SETTINGS_OWNER_LOCK_FILE_NAME,
+            SETTINGS_OWNER_ENDPOINT_FILE_NAME,
+        ] {
+            let rows: Vec<&InstanceArtifact> = INSTANCE_ARTIFACTS
+                .iter()
+                .filter(|row| row.name == name)
+                .collect();
+            assert_eq!(rows.len(), 1, "{name} must have exactly one row");
+            assert_eq!(rows[0].kind, ArtifactKind::File);
+            assert_eq!(rows[0].disposition, Disposition::Ignore);
+        }
+        // Neighbouring rows survive unchanged: the transactional lock and the
+        // retired lock are still Ignore rows of their own.
+        for name in [
+            SETTINGS_LOCK_FILE_NAME,
+            SETTINGS_RETIRED_LOCK_NAME,
+            "sessions.json",
+        ] {
+            assert!(ignore_rows().iter().any(|row| row.name == name), "{name}");
+        }
+    }
 }
diff --git a/src-tauri/src/config/settings.rs b/src-tauri/src/config/settings.rs
index f832ed8c8..2dba9ddcb 100644
--- a/src-tauri/src/config/settings.rs
+++ b/src-tauri/src/config/settings.rs
@@ -13,6 +13,7 @@ use crate::config::instance_artifacts::{
     BLOCKING_MENUS_REMOTE_CHECK_FILE_NAME, BLOCKING_MENUS_REMOTE_FILE_NAME,
     BLOCKING_MENUS_SHIPPED_FILE_NAME, SETTINGS_BACKUP_PREFIX, SETTINGS_BACKUP_SUFFIX,
     SETTINGS_FILE_NAME, SETTINGS_LOCAL_OVERRIDE_FILE_NAME, SETTINGS_LOCK_FILE_NAME,
+    SETTINGS_OWNER_LOCK_FILE_NAME,
 };
 use crate::config::local_overlay::{DerivedIdClosure, LocalSettingsOverlay, PairedLeaf};
 use crate::config::placeholders::AC_PLACEHOLDER_TOKENS;
@@ -8013,7 +8014,16 @@ fn read_agents_instance_file_labeled(
         Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
         Err(e) => return Err(format!("could not read {}: {e}", label.display())),
     };
-    let map = match serde_json::from_str::<Value>(&contents) {
+    parse_agents_instance_contents(&contents, label).map(Some)
+}
+
+/// The decode half of `read_agents_instance_file_labeled`, shared with the owner's
+/// single capture so both judge the same bytes the same way.
+fn parse_agents_instance_contents(
+    contents: &str,
+    label: &Path,
+) -> Result<Map<String, Value>, String> {
+    let map = match serde_json::from_str::<Value>(contents) {
         Ok(Value::Object(map)) => map,
         Ok(_) => return Err(format!("{} is not a JSON object", label.display())),
         Err(e) => return Err(format!("{} is not valid JSON: {e}", label.display())),
@@ -8030,7 +8040,7 @@ fn read_agents_instance_file_labeled(
             )
         })?;
     }
-    Ok(Some(map))
+    Ok(map)
 }
 
 /// #2716 (B3) - the groups `object` actually holds, key by key.
@@ -8196,9 +8206,21 @@ fn merge_agents_instance_file_with_capture(
     capture: Option<&mut Option<StartupAgentRows>>,
 ) -> AgentsLayerState {
     let path = agents_instance_path(settings_path);
+    let read = read_agents_instance_file_labeled(&agents_instance_path(io_path), &path);
+    merge_agents_instance_groups(path, read, base, capture)
+}
+
+/// The merge half of `merge_agents_instance_file_with_capture`, over a read the
+/// caller already made, so one capture never reads the agents file twice.
+fn merge_agents_instance_groups(
+    path: PathBuf,
+    read: AgentsInstanceRead,
+    base: &mut Value,
+    capture: Option<&mut Option<StartupAgentRows>>,
+) -> AgentsLayerState {
     let mut sidecar_owner = false;
     let mut sidecar_absent = false;
-    let state = match read_agents_instance_file_labeled(&agents_instance_path(io_path), &path) {
+    let state = match read {
         Ok(None) => {
             sidecar_absent = true;
             AgentsLayerState::Readable
@@ -8721,20 +8743,76 @@ fn parse_settings_with_startup(
     io_path: Option<&Path>,
     settings_path: Option<&Path>,
     projection: Option<&Map<String, Value>>,
-    mut startup: Option<&mut Option<StartupAgentRows>>,
+    startup: Option<&mut Option<StartupAgentRows>>,
 ) -> Result<(AppSettings, bool), String> {
-    let mut value: Value = serde_json::from_str(contents)
+    let value: Value = serde_json::from_str(contents)
         .map_err(|e| format!("Failed to parse settings file: {}", e))?;
+    // One read of the agents file serves both the physical-group census and the
+    // layer-30 merge; the merge label is the settings path, as before.
+    let agents = io_path.map(|io| {
+        read_agents_instance_file_labeled(
+            &agents_instance_path(io),
+            &agents_instance_path(settings_path.unwrap_or(io)),
+        )
+    });
+    let (mut settings, legacy_profiles, agents_layer) = compose_settings_parts(
+        value,
+        source,
+        settings_path,
+        projection,
+        agents,
+        |value| match io_path {
+            Some(p) => LocalSettingsOverlay::load_and_merge_paired(
+                p,
+                value,
+                OVERLAY_INELIGIBLE_DISK_KEYS,
+                OVERLAY_INELIGIBLE_LEGACY_KEYS,
+                OVERLAY_DERIVED_ID_CLOSURES,
+                CONFIGURATION_COMMAND_PAIRS,
+            ),
+            None => LocalSettingsOverlay::default(),
+        },
+        startup,
+    )?;
+    if let (Some(p), Some(state)) = (settings_path, agents_layer) {
+        set_agents_layer_state(&mut settings, p, state);
+    }
+    repair_loaded_response_close_settings(&mut settings, source);
+    Ok((settings, legacy_profiles))
+}
+
+/// One read of the agents file: `Ok(None)` is verified absence.
+type AgentsInstanceRead = Result<Option<Map<String, Value>>, String>;
+
+/// Settings, the legacy-profiles flag and the agents layer state, unregistered.
+type ComposedSettings = (AppSettings, bool, Option<AgentsLayerState>);
+
+/// The pure composition core of `parse_settings_with_startup`: projection, the
+/// six-field decode, agents groups, carrier and physical identity, the overlay
+/// over the same lists and pairs, serde, order and id hashes. It performs no IO
+/// itself; the agents read and the overlay merge are supplied by the caller, so
+/// the owner's single capture reuses this exact sequence over captured bytes.
+/// `merge_label` is the settings path that enables the agents merge and the
+/// undecodable-overlay fallback. The agents layer state is returned, not
+/// registered, so the caller decides whether a process-wide registry changes.
+fn compose_settings_parts(
+    mut value: Value,
+    source: &str,
+    merge_label: Option<&Path>,
+    projection: Option<&Map<String, Value>>,
+    agents: Option<AgentsInstanceRead>,
+    overlay_merge: impl FnOnce(&mut Value) -> LocalSettingsOverlay,
+    mut startup: Option<&mut Option<StartupAgentRows>>,
+) -> Result<ComposedSettings, String> {
+    let settings_path = merge_label;
     if let (Some(object), Some(projection)) = (value.as_object_mut(), projection) {
         replace_project_projection(object, projection);
     }
     let legacy_profiles = value.as_object().is_some_and(legacy_profiles_shape_present);
-    let physical_groups = io_path
-        .and_then(|path| {
-            read_agents_instance_file(&agents_instance_path(path))
-                .ok()
-                .flatten()
-        })
+    let physical_groups = agents
+        .as_ref()
+        .and_then(|read| read.as_ref().ok())
+        .and_then(Option::as_ref)
         .map(|groups| {
             groups
                 .keys()
@@ -8747,8 +8825,13 @@ fn parse_settings_with_startup(
     // merged into the raw base object here: after the detection above, so
     // `legacy_profiles` stays a property of the settings file, and before the
     // `pre_merge` clone, so the undecodable fallback carries its groups.
-    let agents_layer = io_path.zip(settings_path).map(|(io, label)| {
-        merge_agents_instance_file_with_capture(io, label, &mut value, startup.as_deref_mut())
+    let agents_layer = settings_path.zip(agents).map(|(label, read)| {
+        merge_agents_instance_groups(
+            agents_instance_path(label),
+            read,
+            &mut value,
+            startup.as_deref_mut(),
+        )
     });
     // #1737 (D21): the clone must be taken BEFORE the merge, because the merge is
     // in place and the overlay's presence is not known until it returns. Gated on
@@ -8758,17 +8841,7 @@ fn parse_settings_with_startup(
     // #1737 (D5): the merge point, after the legacy-shape detection and before the
     // project decode. Merging after the detection keeps the `legacy_profiles` flag
     // a property of the base file alone and makes the B7 cross-key edge impossible.
-    let mut overlay = match io_path {
-        Some(p) => LocalSettingsOverlay::load_and_merge_paired(
-            p,
-            &mut value,
-            OVERLAY_INELIGIBLE_DISK_KEYS,
-            OVERLAY_INELIGIBLE_LEGACY_KEYS,
-            OVERLAY_DERIVED_ID_CLOSURES,
-            CONFIGURATION_COMMAND_PAIRS,
-        ),
-        None => LocalSettingsOverlay::default(),
-    };
+    let mut overlay = overlay_merge(&mut value);
     if let (Some(object), Some(base)) = (value.as_object_mut(), &pre_merge) {
         restore_codex_home_carrier(object, base.get("codingAgentProfiles"));
     }
@@ -8833,11 +8906,7 @@ fn parse_settings_with_startup(
             *startup = None;
         }
     }
-    if let (Some(p), Some(state)) = (settings_path, agents_layer) {
-        set_agents_layer_state(&mut settings, p, state);
-    }
-    repair_loaded_response_close_settings(&mut settings, source);
-    Ok((settings, legacy_profiles))
+    Ok((settings, legacy_profiles, agents_layer))
 }
 
 /// #2015: legacy pre-v2 codingAgentProfiles are no longer migrated. Detection only;
@@ -12244,6 +12313,17 @@ fn read_raw_project_source_labeled_with_limit(
     let Some(bytes) = read_project_source_bytes_labeled_with_limit(path, label, limit)? else {
         return Ok(None);
     };
+    raw_project_source_from_bytes(bytes, path, label, legacy_settings_diagnostic).map(Some)
+}
+
+/// The decode half of the raw project-source reader, shared with the owner's
+/// single capture so selection judges captured bytes exactly like read ones.
+fn raw_project_source_from_bytes(
+    bytes: Vec<u8>,
+    path: &Path,
+    label: &Path,
+    legacy_settings_diagnostic: bool,
+) -> Result<RawProjectSource, SettingsSaveError> {
     let value = crate::path_identity::parse_json_no_duplicates(&bytes).map_err(|reason| {
         let mut error = project_migration_error_labeled(
             path,
@@ -12267,7 +12347,7 @@ fn read_raw_project_source_labeled_with_limit(
             "project_paths_source_not_object",
         ));
     };
-    Ok(Some(RawProjectSource { object, bytes }))
+    Ok(RawProjectSource { object, bytes })
 }
 
 #[derive(Debug)]
@@ -12311,6 +12391,24 @@ fn select_project_source_at_labeled_path_with_limit(
         limit,
         false,
     )?;
+    select_project_source_from_raw(
+        legacy,
+        destination,
+        label,
+        &destination_path,
+        &destination_label,
+    )
+}
+
+/// The pure selector over both raw sources: the destination wins, and a non-empty
+/// legacy projection that differs from it is a conflict, never a silent choice.
+fn select_project_source_from_raw(
+    legacy: Option<RawProjectSource>,
+    destination: Option<RawProjectSource>,
+    label: &Path,
+    destination_path: &Path,
+    destination_label: &Path,
+) -> Result<SelectedProjectSource, SettingsSaveError> {
     let legacy_projection = legacy
         .as_ref()
         .map(|source| raw_project_projection(&source.object))
@@ -12320,7 +12418,7 @@ fn select_project_source_at_labeled_path_with_limit(
         Some(source) => {
             let projection = raw_project_projection(&source.object);
             if !legacy_projection.is_empty() && legacy_projection != projection {
-                return Err(project_migration_error_labeled(&destination_path, &destination_label, SettingsSaveStage::ProjectPathsRefreshDiskGate,
+                return Err(project_migration_error_labeled(destination_path, destination_label, SettingsSaveStage::ProjectPathsRefreshDiskGate,
                     &format!("project_paths_conflict; source={}; destination={}; preserve both files and reconcile raw projections to equality", label.display(), destination_label.display())));
             }
             projection
@@ -13011,6 +13109,7 @@ fn settings_lock_debug_emit(event: &SettingsLockDebugEvent) {
 impl SettingsLockDebugAttempt {
     fn prepare(
         settings_path: &Path,
+        lock_name: &str,
         operation: &'static str,
         caller: &'static std::panic::Location<'static>,
     ) -> Option<Self> {
@@ -13034,7 +13133,7 @@ impl SettingsLockDebugAttempt {
         let lock_path = settings_path
             .parent()
             .unwrap_or_else(|| Path::new(""))
-            .join(SETTINGS_LOCK_FILE_NAME);
+            .join(lock_name);
         let (sidecar, sidecar_truncated) = settings_lock_debug_path(&lock_path);
         let debug = Self {
             context: Arc::new(SettingsLockDebugContext {
@@ -13142,14 +13241,26 @@ impl SettingsFileLock {
         settings_path: &Path,
         timeout: std::time::Duration,
         operation: &'static str,
+    ) -> Result<Self, SettingsSaveError> {
+        Self::acquire_named(settings_path, SETTINGS_LOCK_FILE_NAME, timeout, operation)
+    }
+
+    /// The same acquisition over another stable sidecar beside the settings file.
+    #[track_caller]
+    fn acquire_named(
+        settings_path: &Path,
+        lock_name: &str,
+        timeout: std::time::Duration,
+        operation: &'static str,
     ) -> Result<Self, SettingsSaveError> {
         let mut debug = SettingsLockDebugAttempt::prepare(
             settings_path,
+            lock_name,
             operation,
             std::panic::Location::caller(),
         );
         // Inner failure drops every opened file before a diagnostic is enqueued.
-        let file = match Self::acquire_file(settings_path, timeout, &mut debug) {
+        let file = match Self::acquire_file(settings_path, lock_name, timeout, &mut debug) {
             Ok(file) => file,
             Err(error) => {
                 if let Some(debug) = &debug {
@@ -13166,6 +13277,7 @@ impl SettingsFileLock {
 
     fn acquire_file(
         settings_path: &Path,
+        lock_name: &str,
         timeout: std::time::Duration,
         debug: &mut Option<SettingsLockDebugAttempt>,
     ) -> Result<std::fs::File, SettingsSaveError> {
@@ -13196,7 +13308,7 @@ impl SettingsFileLock {
                 SettingsSaveLegacyOutward::SettingsLockUnavailable,
             )
         })?;
-        let lock_path = parent.join(SETTINGS_LOCK_FILE_NAME);
+        let lock_path = parent.join(lock_name);
         let mut options = std::fs::OpenOptions::new();
         options.read(true).write(true).create(true);
         #[cfg(unix)]
@@ -14936,10 +15048,1109 @@ fn save_settings_to_path(settings: &AppSettings, path: &Path) -> Result<(), Sett
     write_value_atomic(&value, path).map(|_| ())
 }
 
-pub type SettingsState = Arc<RwLock<AppSettings>>;
+// ---------------------------------------------------------------------------
+// #2994 (E2992 P41) - the settings owner
+// ---------------------------------------------------------------------------
+//
+// One authoritative `AppSettings` lives in the owner's `RwLock`. Until P50 the
+// product runs the owner in `StagedCompatibility`: bootstrap, CLI and persistence
+// keep their current paths through the forwarding methods below, and managed
+// operations stay disabled. `Active` owners hold the lifecycle lease and run the
+// complete operation contract; only tests construct them before P46.
+
+/// Lifecycle availability of a settings owner. Owner metadata, changed only under
+/// its control mutex. `Invalid` is sticky: it denies current reads, mutation and
+/// authorization, and leaves only diagnostics and a validated close or restart.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SettingsAvailability {
+    Starting,
+    Ready,
+    Invalid,
+    Draining,
+    Closed,
+}
+
+/// Errors of the owner contract. Messages carry stages and reason codes only,
+/// never payloads, tokens or paths.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SettingsOwnerError {
+    /// The owner is not `Ready`, or admission is sealed.
+    Unavailable(SettingsAvailability),
+    /// A try-acquisition found the state lock held.
+    Busy,
+    /// Managed operations are not enabled on this owner (StagedCompatibility).
+    OperationsNotEnabled,
+    /// The lifecycle lock is held by another owner past the wait deadline.
+    LifecycleTimedOut,
+    /// The lifecycle lock or its directory identity could not be established.
+    /// Never evidence that the authority is free.
+    Lifecycle(String),
+    /// The transactional lock could not be taken for an operation.
+    TransactionLock(String),
+    /// A source could not be captured; IO, identity or malformation, never absence.
+    Capture(String),
+    /// The captured sources do not compose under the requested policy.
+    Compose(String),
+    /// The blocking worker could not be joined.
+    Worker(String),
+}
+
+impl std::fmt::Display for SettingsOwnerError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Unavailable(availability) => {
+                write!(formatter, "settings owner unavailable ({availability:?})")
+            }
+            Self::Busy => formatter.write_str("settings owner state lock is busy"),
+            Self::OperationsNotEnabled => {
+                formatter.write_str("settings owner managed operations are not enabled")
+            }
+            Self::LifecycleTimedOut => {
+                formatter.write_str("settings owner lifecycle lock is held by another owner")
+            }
+            Self::Lifecycle(reason) => write!(formatter, "settings owner lifecycle: {reason}"),
+            Self::TransactionLock(reason) => {
+                write!(formatter, "settings owner transaction lock: {reason}")
+            }
+            Self::Capture(reason) => write!(formatter, "settings owner capture: {reason}"),
+            Self::Compose(reason) => write!(formatter, "settings owner composition: {reason}"),
+            Self::Worker(reason) => write!(formatter, "settings owner worker: {reason}"),
+        }
+    }
+}
+
+impl std::error::Error for SettingsOwnerError {}
+
+/// Stage and reason code of a settings save error, without its paths.
+fn owner_save_error_text(error: &SettingsSaveError) -> String {
+    let reason = match &error.cause {
+        SettingsSaveCause::Io(source) => io_error_kind_as_str(source.kind()),
+        SettingsSaveCause::Json(source) => json_error_category_as_str(source.classify()),
+        SettingsSaveCause::Semantic(reason) => reason.as_str(),
+    };
+    format!("{error} ({:?}: {reason})", error.stage)
+}
+
+fn owner_transaction_error(error: SettingsSaveError) -> SettingsOwnerError {
+    SettingsOwnerError::TransactionLock(owner_save_error_text(&error))
+}
+
+/// Identity the owner reports upward: no transport, no `AppHandle`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SettingsOwnerIdentity {
+    pub instance_id: uuid::Uuid,
+    pub epoch: u64,
+    pub availability: SettingsAvailability,
+    pub managed_operations_enabled: bool,
+}
+
+/// Project retention captured from RAM. A `source_error` means the lists are
+/// unknown, never empty: no consumer may purge on it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SettingsRetention {
+    pub active: Vec<String>,
+    pub archived: Vec<String>,
+    pub source_error: Option<String>,
+}
+
+/// Whether an operation may already have touched a persistent source.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OperationEffect {
+    None,
+    Possible,
+}
+
+/// How an operation ended. `PartialOrUnknown` is an honest error: never a
+/// success and never a rollback.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OperationOutcome {
+    Committed,
+    Rejected,
+    PartialOrUnknown,
+}
+
+/// The publishers a locked operation may run. Each one flips the effect to
+/// `Possible` before it starts.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SettingsPublisher {
+    PreflightBackup,
+    ProjectPathsMigration,
+    ProjectPathsBackup,
+    ProjectPathsCleanup,
+    BlockingMenusExport,
+    AgentsExport,
+    AgentsSave,
+    SettingsSave,
+    AuxCache,
+    Stamp,
+}
+
+/// Composition policy. Startup keeps overlay and agents rejections as metadata;
+/// recovery requires every affected source to be valid. Both keep legitimate
+/// absence and neither turns IO or a base decode failure into defaults.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CapturePolicy {
+    Startup,
+    Recovery,
+}
+
+/// The result an operation returns once the owner has adopted (or refused) the
+/// effective reality.
+#[derive(Debug)]
+pub struct SettingsOperationReport<T> {
+    pub outcome: OperationOutcome,
+    pub value: Option<T>,
+    pub error: Option<String>,
+    pub epoch: u64,
+    pub availability: SettingsAvailability,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SettingsOwnerStage {
+    StagedCompatibility,
+    Active,
+}
+
+#[derive(Debug)]
+struct SettingsOwnerControl {
+    availability: SettingsAvailability,
+    epoch: u64,
+    permits: usize,
+    operations: usize,
+    sealed: bool,
+    managed_operations_enabled: bool,
+}
+
+/// Counters for tests and DEBUG diagnostics; no payloads.
+#[derive(Debug, Default)]
+struct SettingsOwnerCounters {
+    transaction_lock_attempts: AtomicU64,
+    captures: AtomicU64,
+    source_reads: AtomicU64,
+    operations: AtomicU64,
+}
+
+/// The lifecycle lease: an exclusive lock on the stable owner sidecar inside a
+/// retained physical directory. Dropping it unlocks; the sidecar is never
+/// unlinked.
+struct SettingsOwnerLease {
+    _lock: SettingsFileLock,
+    retained: crate::path_identity::RetainedDirectory,
+    settings_path: PathBuf,
+}
+
+impl SettingsOwnerLease {
+    fn acquire(
+        settings_path: &Path,
+        wait: std::time::Duration,
+    ) -> Result<Self, SettingsOwnerError> {
+        let identity = |reason: &str| SettingsOwnerError::Lifecycle(reason.to_string());
+        let parent = settings_path
+            .parent()
+            .ok_or_else(|| identity("missing_parent_directory"))?;
+        let leaf = settings_path
+            .file_name()
+            .ok_or_else(|| identity("missing_file_name"))?;
+        // The chain check stops at the first link, so a `..` after a tolerated
+        // link (macOS `/var`) would otherwise be resolved by canonicalize.
+        if parent
+            .components()
+            .any(|component| component == std::path::Component::ParentDir)
+        {
+            return Err(identity("lock_identity_rejected"));
+        }
+        match crate::path_identity::verify_component_chain_reason(parent) {
+            Ok(()) | Err(crate::path_identity::PathRule::ComponentIsLink) => {}
+            Err(_) => return Err(identity("lock_identity_rejected")),
+        }
+        let physical_parent =
+            std::fs::canonicalize(parent).map_err(|_| identity("lock_identity_rejected"))?;
+        let retained = crate::path_identity::retain_directory(&physical_parent)
+            .map_err(|_| identity("lock_identity_rejected"))?;
+        let physical = retained.identity().canonical_path.join(leaf);
+        let lock = SettingsFileLock::acquire_named(
+            &physical,
+            SETTINGS_OWNER_LOCK_FILE_NAME,
+            wait,
+            "settings_owner_lifecycle",
+        )
+        .map_err(|error| match error.cause {
+            SettingsSaveCause::Semantic(SettingsSaveReason::LockTimedOut) => {
+                SettingsOwnerError::LifecycleTimedOut
+            }
+            _ => SettingsOwnerError::Lifecycle(owner_save_error_text(&error)),
+        })?;
+        retained
+            .verify_current()
+            .map_err(|_| identity("lock_identity_rejected"))?;
+        Ok(Self {
+            _lock: lock,
+            retained,
+            settings_path: physical,
+        })
+    }
+
+    fn verify(&self) -> Result<(), SettingsOwnerError> {
+        self.retained
+            .verify_current()
+            .map_err(|_| SettingsOwnerError::Lifecycle("lock_identity_rejected".to_string()))
+    }
+}
+
+/// The single settings owner. `SettingsState` is `Arc<SettingsOwner>`.
+pub struct SettingsOwner {
+    settings: Arc<RwLock<AppSettings>>,
+    stage: SettingsOwnerStage,
+    instance_id: uuid::Uuid,
+    control: std::sync::Mutex<SettingsOwnerControl>,
+    released: std::sync::Condvar,
+    lease: std::sync::Mutex<Option<SettingsOwnerLease>>,
+    counters: SettingsOwnerCounters,
+}
+
+impl std::fmt::Debug for SettingsOwner {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("SettingsOwner")
+            .field("stage", &self.stage)
+            .field("instance_id", &self.instance_id)
+            .field("control", &*self.control())
+            .finish_non_exhaustive()
+    }
+}
+
+/// Admission of one read permit; releasing it wakes a draining owner.
+struct SettingsPermitAdmission<'a> {
+    owner: &'a SettingsOwner,
+}
+
+impl Drop for SettingsPermitAdmission<'_> {
+    fn drop(&mut self) {
+        self.owner.release_permit();
+    }
+}
+
+/// Admission of one managed operation. It is released on every path, unwinding
+/// included, which is what lets a drain terminate.
+struct SettingsOperationAdmission<'a> {
+    owner: &'a SettingsOwner,
+}
+
+impl Drop for SettingsOperationAdmission<'_> {
+    fn drop(&mut self) {
+        self.owner.finish_operation();
+    }
+}
+
+/// Outcome, builder value and error of one classified operation.
+type OperationParts<T> = (OperationOutcome, Option<T>, Option<String>);
+
+/// A borrowed, validated read of the current settings.
+pub struct SettingsReadGuard<'a> {
+    guard: tokio::sync::RwLockReadGuard<'a, AppSettings>,
+    _admission: SettingsPermitAdmission<'a>,
+}
+
+impl std::ops::Deref for SettingsReadGuard<'_> {
+    type Target = AppSettings;
+
+    fn deref(&self) -> &AppSettings {
+        &self.guard
+    }
+}
+
+/// An owned, validated read permit: the state guard plus the owner it belongs to.
+/// A PTY route write keeps it until the write ends.
+pub struct SettingsReadPermit {
+    guard: Option<tokio::sync::OwnedRwLockReadGuard<AppSettings>>,
+    owner: Arc<SettingsOwner>,
+}
+
+impl SettingsReadPermit {
+    pub fn owner(&self) -> &Arc<SettingsOwner> {
+        &self.owner
+    }
+}
+
+impl std::ops::Deref for SettingsReadPermit {
+    type Target = AppSettings;
+
+    fn deref(&self) -> &AppSettings {
+        self.guard
+            .as_deref()
+            .expect("a settings read permit holds its guard until drop")
+    }
+}
+
+impl Drop for SettingsReadPermit {
+    fn drop(&mut self) {
+        drop(self.guard.take());
+        self.owner.release_permit();
+    }
+}
+
+/// Proof that an operation holds the owner's lifecycle authority and the current
+/// transactional lock. Constructed only inside the owner; the locked halves take
+/// it instead of reacquiring the transactional lock.
+pub struct SettingsOperationGuard<'a> {
+    lease: &'a SettingsOwnerLease,
+    transaction: ProjectPathsWriteTransaction,
+    counters: &'a SettingsOwnerCounters,
+    effect: OperationEffect,
+}
+
+impl SettingsOperationGuard<'_> {
+    /// The physical settings path under the retained directory.
+    pub fn settings_path(&self) -> &Path {
+        &self.transaction.path
+    }
+
+    pub fn effect(&self) -> OperationEffect {
+        self.effect
+    }
+
+    /// Runs one publisher. The effect becomes `Possible` BEFORE the publisher
+    /// starts, so a failure from here on classifies as `PartialOrUnknown`.
+    pub fn publish<R>(
+        &mut self,
+        publisher: SettingsPublisher,
+        run: impl FnOnce(&Path) -> Result<R, String>,
+    ) -> Result<R, String> {
+        self.effect = OperationEffect::Possible;
+        log::debug!("[settings-owner] publisher start: {publisher:?}");
+        let result = run(&self.transaction.path);
+        log::debug!(
+            "[settings-owner] publisher end: {publisher:?} ok={}",
+            result.is_ok()
+        );
+        result
+    }
+}
+
+/// One captured source: verified absence, or its bytes with the identity witness
+/// of the object they were read from.
+#[derive(Debug, Clone)]
+pub(crate) enum CapturedSource {
+    Absent,
+    Present {
+        bytes: Vec<u8>,
+        witness: crate::path_identity::VerifiedPathIdentity,
+    },
+}
+
+impl CapturedSource {
+    fn bytes(&self) -> Option<&[u8]> {
+        match self {
+            Self::Absent => None,
+            Self::Present { bytes, .. } => Some(bytes),
+        }
+    }
+
+    /// A short, non-reversible witness label for DEBUG diagnostics.
+    fn witness_label(&self) -> String {
+        match self {
+            Self::Absent => "absent".to_string(),
+            Self::Present { witness, .. } => match witness.content_sha256 {
+                Some(digest) => digest[..4]
+                    .iter()
+                    .map(|byte| format!("{byte:02x}"))
+                    .collect(),
+                None => format!("len{}", witness.metadata.len),
+            },
+        }
+    }
+}
+
+/// The four settings sources, captured once under the transactional lock.
+#[derive(Debug, Clone)]
+pub(crate) struct CapturedSettingsSources {
+    settings_path: PathBuf,
+    base: CapturedSource,
+    project_paths: CapturedSource,
+    agents: CapturedSource,
+    overlay: CapturedSource,
+}
+
+const SETTINGS_CAPTURE_LIMIT: usize = 16 * 1024 * 1024;
+
+fn capture_source_file(
+    path: &Path,
+    counters: &SettingsOwnerCounters,
+) -> Result<CapturedSource, SettingsOwnerError> {
+    counters.source_reads.fetch_add(1, Ordering::Relaxed);
+    let name = path
+        .file_name()
+        .map(|name| name.to_string_lossy().into_owned())
+        .unwrap_or_default();
+    match std::fs::symlink_metadata(path) {
+        Ok(_) => {}
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Ok(CapturedSource::Absent);
+        }
+        Err(error) => {
+            let kind = io_error_kind_as_str(error.kind());
+            return Err(SettingsOwnerError::Capture(format!("{name}: {kind}")));
+        }
+    }
+    // A source that vanishes after the metadata probe is a race, not absence.
+    crate::path_identity::read_bounded_regular(path, SETTINGS_CAPTURE_LIMIT)
+        .map(|(bytes, witness)| CapturedSource::Present { bytes, witness })
+        .map_err(|reason| SettingsOwnerError::Capture(format!("{name}: {reason}")))
+}
+
+// Test seam: the next owner capture on this thread panics, so the panic is
+// raised outside the builder while the operation's guards are held.
+#[cfg(test)]
+thread_local! {
+    static OWNER_CAPTURE_PANIC: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
+}
+
+/// Captures base, project-paths, agents and overlay exactly once while the guard
+/// holds the transactional lock. No directory creation, migration or write.
+fn capture_sources_locked(
+    guard: &SettingsOperationGuard<'_>,
+) -> Result<CapturedSettingsSources, SettingsOwnerError> {
+    guard.counters.captures.fetch_add(1, Ordering::Relaxed);
+    #[cfg(test)]
+    OWNER_CAPTURE_PANIC.with(|armed| {
+        if armed.replace(false) {
+            panic!("injected settings owner capture panic");
+        }
+    });
+    let path = guard.settings_path();
+    let captured = CapturedSettingsSources {
+        settings_path: path.to_path_buf(),
+        base: capture_source_file(path, guard.counters)?,
+        project_paths: capture_source_file(&project_paths_path(path), guard.counters)?,
+        agents: capture_source_file(&agents_instance_path(path), guard.counters)?,
+        overlay: capture_source_file(
+            &path.with_file_name(SETTINGS_LOCAL_OVERRIDE_FILE_NAME),
+            guard.counters,
+        )?,
+    };
+    guard.lease.verify()?;
+    log::debug!(
+        "[settings-owner] captured base={} project={} agents={} overlay={}",
+        captured.base.witness_label(),
+        captured.project_paths.witness_label(),
+        captured.agents.witness_label(),
+        captured.overlay.witness_label()
+    );
+    Ok(captured)
+}
+
+fn recover_composition(
+    captured: &CapturedSettingsSources,
+) -> Result<AppSettings, SettingsOwnerError> {
+    compose_captured_sources(captured, CapturePolicy::Recovery)
+}
+
+/// Composes captured bytes through the same selector and parser core the loaders
+/// use. No second read, no defaults for a present source that fails.
+fn compose_captured_sources(
+    captured: &CapturedSettingsSources,
+    policy: CapturePolicy,
+) -> Result<AppSettings, SettingsOwnerError> {
+    let path = captured.settings_path.as_path();
+    let destination_path = project_paths_path(path);
+    let selection_error = |error: SettingsSaveError| {
+        SettingsOwnerError::Compose(format!("selection: {}", owner_save_error_text(&error)))
+    };
+    let raw = |source: &CapturedSource, label: &Path, legacy: bool| {
+        source
+            .bytes()
+            .map(|bytes| raw_project_source_from_bytes(bytes.to_vec(), label, label, legacy))
+            .transpose()
+    };
+    let legacy = raw(&captured.base, path, true).map_err(selection_error)?;
+    let destination = raw(&captured.project_paths, &destination_path, false);
+    let destination = destination.map_err(selection_error)?;
+    let selected = select_project_source_from_raw(
+        legacy,
+        destination,
+        path,
+        &destination_path,
+        &destination_path,
+    )
+    .map_err(selection_error)?;
+    let agents_label = agents_instance_path(path);
+    let agents = match captured.agents.bytes() {
+        None => Ok(None),
+        Some(bytes) => std::str::from_utf8(bytes)
+            .map_err(|_| format!("{} is not valid UTF-8", agents_label.display()))
+            .and_then(|contents| parse_agents_instance_contents(contents, &agents_label))
+            .map(Some),
+    };
+    if policy == CapturePolicy::Recovery && agents.is_err() {
+        let reason = "agents source is invalid during recovery";
+        return Err(SettingsOwnerError::Compose(reason.to_string()));
+    }
+    let source = path.to_string_lossy();
+    // Only a verified-absent base composes over serialized defaults.
+    let value = match &selected.legacy {
+        Some(legacy) => Value::Object(legacy.object.clone()),
+        None => serde_json::to_value(AppSettings::default())
+            .map_err(|error| SettingsOwnerError::Compose(format!("defaults: {error}")))?,
+    };
+    let overlay_bytes = captured.overlay.bytes();
+    let (mut settings, _legacy_profiles, agents_layer) = compose_settings_parts(
+        value,
+        &source,
+        Some(path),
+        Some(&selected.projection),
+        Some(agents),
+        |value| match overlay_bytes {
+            Some(bytes) => LocalSettingsOverlay::from_overlay_bytes_paired(
+                value,
+                bytes,
+                OVERLAY_INELIGIBLE_DISK_KEYS,
+                OVERLAY_INELIGIBLE_LEGACY_KEYS,
+                OVERLAY_DERIVED_ID_CLOSURES,
+                CONFIGURATION_COMMAND_PAIRS,
+            ),
+            None => LocalSettingsOverlay::default(),
+        },
+        None,
+    )
+    .map_err(|_| SettingsOwnerError::Compose("base settings do not decode".to_string()))?;
+    let overlay_rejected = settings.local_overlay_state.rejection().is_some();
+    if policy == CapturePolicy::Recovery && overlay_rejected {
+        let reason = "overlay rejected during recovery";
+        return Err(SettingsOwnerError::Compose(reason.to_string()));
+    }
+    if let Some(state) = agents_layer {
+        settings.agents_layer = state;
+    }
+    repair_loaded_response_close_settings(&mut settings, &source);
+    Ok(settings)
+}
+
+impl SettingsOwner {
+    /// StagedCompatibility owner around already-loaded settings: the P40
+    /// constructor path. Ready, with managed operations disabled.
+    fn staged(settings: AppSettings) -> Self {
+        Self::with_stage(
+            settings,
+            SettingsOwnerStage::StagedCompatibility,
+            SettingsAvailability::Ready,
+            None,
+        )
+    }
+
+    fn with_stage(
+        settings: AppSettings,
+        stage: SettingsOwnerStage,
+        availability: SettingsAvailability,
+        lease: Option<SettingsOwnerLease>,
+    ) -> Self {
+        Self {
+            settings: Arc::new(RwLock::new(settings)),
+            stage,
+            instance_id: uuid::Uuid::new_v4(),
+            control: std::sync::Mutex::new(SettingsOwnerControl {
+                availability,
+                epoch: 0,
+                permits: 0,
+                operations: 0,
+                sealed: false,
+                managed_operations_enabled: stage == SettingsOwnerStage::Active,
+            }),
+            released: std::sync::Condvar::new(),
+            lease: std::sync::Mutex::new(lease),
+            counters: SettingsOwnerCounters::default(),
+        }
+    }
+
+    /// Opens an Active owner: lifecycle authority first, then the transactional
+    /// lock, one capture and a Startup composition before `Ready`. Another owner
+    /// holding the authority makes this wait up to `wait`, then fail.
+    pub fn open_active(
+        settings_path: &Path,
+        wait: std::time::Duration,
+    ) -> Result<Arc<Self>, SettingsOwnerError> {
+        let started = std::time::Instant::now();
+        let lease = SettingsOwnerLease::acquire(settings_path, wait)?;
+        let mut owner = Self::with_stage(
+            AppSettings::default(),
+            SettingsOwnerStage::Active,
+            SettingsAvailability::Starting,
+            Some(lease),
+        );
+        let initial = {
+            let lease = owner.lease_guard();
+            let lease = lease.as_ref().ok_or(SettingsOwnerError::Unavailable(
+                SettingsAvailability::Closed,
+            ))?;
+            let guard = owner.operation_guard(lease)?;
+            let captured = capture_sources_locked(&guard)?;
+            compose_captured_sources(&captured, CapturePolicy::Startup)?
+        };
+        // Not shared yet: the initial value replaces the placeholder without a lock.
+        owner.settings = Arc::new(RwLock::new(initial));
+        {
+            let mut control = owner.control();
+            control.availability = SettingsAvailability::Ready;
+            control.epoch = 1;
+        }
+        log::debug!(
+            "[settings-owner] {} ready in {:?}",
+            owner.instance_id,
+            started.elapsed()
+        );
+        Ok(Arc::new(owner))
+    }
+
+    fn control(&self) -> std::sync::MutexGuard<'_, SettingsOwnerControl> {
+        self.control
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    fn lease_guard(&self) -> std::sync::MutexGuard<'_, Option<SettingsOwnerLease>> {
+        self.lease
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub fn identity(&self) -> SettingsOwnerIdentity {
+        let control = self.control();
+        SettingsOwnerIdentity {
+            instance_id: self.instance_id,
+            epoch: control.epoch,
+            availability: control.availability,
+            managed_operations_enabled: control.managed_operations_enabled,
+        }
+    }
+
+    pub fn availability(&self) -> SettingsAvailability {
+        self.control().availability
+    }
+
+    fn admit_permit(&self) -> Result<SettingsPermitAdmission<'_>, SettingsOwnerError> {
+        let mut control = self.control();
+        if control.sealed || control.availability != SettingsAvailability::Ready {
+            return Err(SettingsOwnerError::Unavailable(control.availability));
+        }
+        control.permits += 1;
+        Ok(SettingsPermitAdmission { owner: self })
+    }
+
+    fn release_permit(&self) {
+        let mut control = self.control();
+        control.permits = control.permits.saturating_sub(1);
+        if control.permits == 0 {
+            self.released.notify_all();
+        }
+    }
+
+    fn revalidate(&self) -> Result<(), SettingsOwnerError> {
+        let control = self.control();
+        if control.availability == SettingsAvailability::Ready {
+            Ok(())
+        } else {
+            Err(SettingsOwnerError::Unavailable(control.availability))
+        }
+    }
+
+    fn mark_invalid(&self, stage: &str) {
+        let mut control = self.control();
+        if !matches!(
+            control.availability,
+            SettingsAvailability::Draining | SettingsAvailability::Closed
+        ) {
+            control.availability = SettingsAvailability::Invalid;
+        }
+        let epoch = control.epoch;
+        log::debug!(
+            "[settings-owner] {} invalid at {stage} epoch={epoch}",
+            self.instance_id
+        );
+    }
+
+    /// A borrowed read validated against availability on admission and again
+    /// once the state lock is held.
+    pub async fn current_read(&self) -> Result<SettingsReadGuard<'_>, SettingsOwnerError> {
+        let admission = self.admit_permit()?;
+        let guard = self.settings.read().await;
+        self.revalidate()?;
+        Ok(SettingsReadGuard {
+            guard,
+            _admission: admission,
+        })
+    }
+
+    pub fn try_current_read(&self) -> Result<SettingsReadGuard<'_>, SettingsOwnerError> {
+        let admission = self.admit_permit()?;
+        let guard = self
+            .settings
+            .try_read()
+            .map_err(|_| SettingsOwnerError::Busy)?;
+        self.revalidate()?;
+        Ok(SettingsReadGuard {
+            guard,
+            _admission: admission,
+        })
+    }
+
+    pub fn blocking_current_read(&self) -> Result<SettingsReadGuard<'_>, SettingsOwnerError> {
+        let admission = self.admit_permit()?;
+        let guard = self.settings.blocking_read();
+        self.revalidate()?;
+        Ok(SettingsReadGuard {
+            guard,
+            _admission: admission,
+        })
+    }
+
+    /// An owned permit holding the state guard and this owner.
+    pub fn try_current_read_owned(
+        self: &Arc<Self>,
+    ) -> Result<SettingsReadPermit, SettingsOwnerError> {
+        let admission = self.admit_permit()?;
+        let guard = Arc::clone(&self.settings)
+            .try_read_owned()
+            .map_err(|_| SettingsOwnerError::Busy)?;
+        // The permit now owns the admission and releases it on drop.
+        std::mem::forget(admission);
+        let permit = SettingsReadPermit {
+            guard: Some(guard),
+            owner: Arc::clone(self),
+        };
+        self.revalidate()?;
+        Ok(permit)
+    }
+
+    /// Clones RAM through a validated guard: no filesystem, no transactional lock.
+    pub async fn current_snapshot(&self) -> Result<AppSettings, SettingsOwnerError> {
+        self.current_read().await.map(|guard| guard.clone())
+    }
+
+    pub fn blocking_current_snapshot(&self) -> Result<AppSettings, SettingsOwnerError> {
+        self.blocking_current_read().map(|guard| guard.clone())
+    }
+
+    fn retention_of(settings: &AppSettings) -> SettingsRetention {
+        SettingsRetention {
+            active: settings.project_paths.clone(),
+            archived: settings.archived_project_paths.clone(),
+            source_error: settings.project_paths_persistence_error.clone(),
+        }
+    }
+
+    /// Retention from RAM, captured before any sessions lock is taken.
+    pub async fn current_retention(&self) -> Result<SettingsRetention, SettingsOwnerError> {
+        self.current_read()
+            .await
+            .map(|guard| Self::retention_of(&guard))
+    }
+
+    pub fn blocking_current_retention(&self) -> Result<SettingsRetention, SettingsOwnerError> {
+        self.blocking_current_read()
+            .map(|guard| Self::retention_of(&guard))
+    }
+
+    fn operation_guard<'a>(
+        &'a self,
+        lease: &'a SettingsOwnerLease,
+    ) -> Result<SettingsOperationGuard<'a>, SettingsOwnerError> {
+        lease.verify()?;
+        let attempts = &self.counters.transaction_lock_attempts;
+        attempts.fetch_add(1, Ordering::Relaxed);
+        let operation = "settings_owner_operation";
+        let transaction = empty_project_transaction(&lease.settings_path, operation);
+        let transaction = transaction.map_err(owner_transaction_error)?;
+        Ok(SettingsOperationGuard {
+            lease,
+            transaction,
+            counters: &self.counters,
+            effect: OperationEffect::None,
+        })
+    }
+
+    fn admit_operation(&self) -> Result<SettingsOperationAdmission<'_>, SettingsOwnerError> {
+        let mut control = self.control();
+        if !control.managed_operations_enabled {
+            return Err(SettingsOwnerError::OperationsNotEnabled);
+        }
+        if control.sealed || control.availability != SettingsAvailability::Ready {
+            return Err(SettingsOwnerError::Unavailable(control.availability));
+        }
+        control.operations += 1;
+        Ok(SettingsOperationAdmission { owner: self })
+    }
+
+    fn finish_operation(&self) {
+        let mut control = self.control();
+        control.operations = control.operations.saturating_sub(1);
+        if control.operations == 0 {
+            self.released.notify_all();
+        }
+    }
+
+    /// Runs one complete synchronous operation: admission, state guard, current
+    /// transactional lock, builder over the captured current settings, publishers,
+    /// adoption of the effective reality, then release. Host effects belong to
+    /// the caller, after this returns. Never nests a runtime.
+    ///
+    /// The admission is released on every path, panics included, so a drain
+    /// always terminates. A panic before the guards are held published nothing
+    /// and is a worker error; a panic while they are held is classified under
+    /// them (see `run_operation`).
+    pub fn with_operation_blocking<T>(
+        &self,
+        name: &'static str,
+        builder: impl FnOnce(&AppSettings, &mut SettingsOperationGuard<'_>) -> Result<T, String>,
+    ) -> Result<SettingsOperationReport<T>, SettingsOwnerError> {
+        let _admission = self.admit_operation()?;
+        let run = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            self.run_operation(name, builder)
+        }));
+        match run {
+            Ok(result) => result,
+            Err(_) => {
+                let reason = "operation panicked before its guards were held";
+                Err(SettingsOwnerError::Worker(reason.to_string()))
+            }
+        }
+    }
+
+    /// The async form: the same body on a blocking worker that retains its
+    /// owner handle. The future must not be awaited while holding any guard.
+    pub async fn with_operation<T, F>(
+        self: &Arc<Self>,
+        name: &'static str,
+        builder: F,
+    ) -> Result<SettingsOperationReport<T>, SettingsOwnerError>
+    where
+        T: Send + 'static,
+        F: FnOnce(&AppSettings, &mut SettingsOperationGuard<'_>) -> Result<T, String>
+            + Send
+            + 'static,
+    {
+        let owner = Arc::clone(self);
+        tokio::task::spawn_blocking(move || owner.with_operation_blocking(name, builder))
+            .await
+            .map_err(|error| SettingsOwnerError::Worker(error.to_string()))?
+    }
+
+    fn run_operation<T>(
+        &self,
+        name: &'static str,
+        builder: impl FnOnce(&AppSettings, &mut SettingsOperationGuard<'_>) -> Result<T, String>,
+    ) -> Result<SettingsOperationReport<T>, SettingsOwnerError> {
+        let operation = self.counters.operations.fetch_add(1, Ordering::Relaxed) + 1;
+        let started = std::time::Instant::now();
+        let mut state = self.settings.blocking_write();
+        let lease = self.lease_guard();
+        let lease = lease.as_ref().ok_or(SettingsOwnerError::Unavailable(
+            SettingsAvailability::Closed,
+        ))?;
+        let mut guard = self.operation_guard(lease)?;
+        // Everything that reads sources, runs the builder or adopts reality runs
+        // under this catch, so a panic anywhere is classified while the state
+        // guard, the lease and the transactional lock are all still held.
+        let body = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            self.operation_body(&mut state, &mut guard, builder)
+        }));
+        let effect = guard.effect;
+        let (outcome, value, error) = match body {
+            Ok(Ok(parts)) => parts,
+            Ok(Err(error)) => return Err(error),
+            Err(_) if effect == OperationEffect::None => {
+                // No publisher ran and RAM was not replaced: nothing to recover.
+                let reason = "operation panicked before any publisher";
+                (OperationOutcome::Rejected, None, Some(reason.to_string()))
+            }
+            Err(_) => {
+                // A source may have changed and reality could not be adopted.
+                self.mark_invalid("panic");
+                let reason = "operation panicked after a publisher";
+                (
+                    OperationOutcome::PartialOrUnknown,
+                    None,
+                    Some(reason.to_string()),
+                )
+            }
+        };
+        drop(guard);
+        drop(state);
+        let control = self.control();
+        log::debug!(
+            "[settings-owner] {} op={operation} {name} {outcome:?}/{effect:?} epoch={} in {:?}",
+            self.instance_id,
+            control.epoch,
+            started.elapsed()
+        );
+        Ok(SettingsOperationReport {
+            outcome,
+            value,
+            error,
+            epoch: control.epoch,
+            availability: control.availability,
+        })
+    }
+
+    /// Capture, builder, classification and adoption, all under the caller's
+    /// guards. Admission errors return before any publisher can run.
+    fn operation_body<T>(
+        &self,
+        state: &mut AppSettings,
+        guard: &mut SettingsOperationGuard<'_>,
+        builder: impl FnOnce(&AppSettings, &mut SettingsOperationGuard<'_>) -> Result<T, String>,
+    ) -> Result<OperationParts<T>, SettingsOwnerError> {
+        // Sources that do not compose at admission reject the operation before
+        // any publisher; RAM keeps the last adopted value.
+        let captured = capture_sources_locked(guard)?;
+        let current = compose_captured_sources(&captured, CapturePolicy::Recovery)?;
+        let built = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            builder(&current, &mut *guard)
+        }));
+        let effect = guard.effect;
+        let (value, error) = match built {
+            Ok(Ok(value)) => (Some(value), None),
+            Ok(Err(error)) => (None, Some(error)),
+            Err(_) => (None, Some("operation panicked".to_string())),
+        };
+        let outcome = if error.is_none() {
+            OperationOutcome::Committed
+        } else if effect == OperationEffect::None {
+            OperationOutcome::Rejected
+        } else {
+            OperationOutcome::PartialOrUnknown
+        };
+        // Rejected changed nothing: keep RAM. A success that published nothing
+        // adopts the composition it already captured, without a second capture.
+        // Otherwise recapture under the same guard and adopt reality, or become
+        // Invalid without saving defaults.
+        let outcome = if outcome == OperationOutcome::Rejected {
+            outcome
+        } else if effect == OperationEffect::None {
+            *state = current;
+            self.control().epoch += 1;
+            outcome
+        } else {
+            let recaptured = capture_sources_locked(guard);
+            match recaptured.and_then(|captured| recover_composition(&captured)) {
+                Ok(effective) => {
+                    *state = effective;
+                    self.control().epoch += 1;
+                    outcome
+                }
+                Err(_) => {
+                    self.mark_invalid("recapture");
+                    OperationOutcome::PartialOrUnknown
+                }
+            }
+        };
+        Ok((outcome, value, error))
+    }
+
+    /// Seals admission, waits for every permit and operation, then releases the
+    /// lifecycle authority LAST. The soft deadline only diagnoses; it never
+    /// detaches or unlocks early.
+    pub fn drain_blocking(&self, soft_deadline: std::time::Duration) {
+        let started = std::time::Instant::now();
+        let mut control = self.control();
+        control.sealed = true;
+        if control.availability != SettingsAvailability::Closed {
+            control.availability = SettingsAvailability::Draining;
+        }
+        let mut reported = false;
+        while control.permits > 0 || control.operations > 0 {
+            let (next, timeout) = self
+                .released
+                .wait_timeout(control, std::time::Duration::from_millis(50))
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            control = next;
+            if timeout.timed_out() && !reported && started.elapsed() >= soft_deadline {
+                reported = true;
+                log::debug!(
+                    "[settings-owner] {} drain past soft deadline permits={} operations={}",
+                    self.instance_id,
+                    control.permits,
+                    control.operations
+                );
+            }
+        }
+        control.availability = SettingsAvailability::Closed;
+        drop(control);
+        let lease = self.lease_guard().take();
+        drop(lease);
+        log::debug!(
+            "[settings-owner] {} closed in {:?}",
+            self.instance_id,
+            started.elapsed()
+        );
+    }
+
+    fn staged_forwarding(&self) {
+        debug_assert_eq!(
+            self.stage,
+            SettingsOwnerStage::StagedCompatibility,
+            "lock forwarding is StagedCompatibility only"
+        );
+    }
+
+    // StagedCompatibility forwarding: the pre-owner RwLock surface, kept only so
+    // consumers not yet migrated compile and behave exactly as before. P50
+    // retires it from the product.
+
+    pub async fn read(&self) -> tokio::sync::RwLockReadGuard<'_, AppSettings> {
+        self.staged_forwarding();
+        self.settings.read().await
+    }
+
+    pub async fn write(&self) -> tokio::sync::RwLockWriteGuard<'_, AppSettings> {
+        self.staged_forwarding();
+        self.settings.write().await
+    }
+
+    pub fn try_read(
+        &self,
+    ) -> Result<tokio::sync::RwLockReadGuard<'_, AppSettings>, tokio::sync::TryLockError> {
+        self.staged_forwarding();
+        self.settings.try_read()
+    }
+
+    pub fn try_write(
+        &self,
+    ) -> Result<tokio::sync::RwLockWriteGuard<'_, AppSettings>, tokio::sync::TryLockError> {
+        self.staged_forwarding();
+        self.settings.try_write()
+    }
+
+    pub fn blocking_read(&self) -> tokio::sync::RwLockReadGuard<'_, AppSettings> {
+        self.staged_forwarding();
+        self.settings.blocking_read()
+    }
+
+    pub fn blocking_write(&self) -> tokio::sync::RwLockWriteGuard<'_, AppSettings> {
+        self.staged_forwarding();
+        self.settings.blocking_write()
+    }
+
+    pub async fn read_owned(self: Arc<Self>) -> tokio::sync::OwnedRwLockReadGuard<AppSettings> {
+        self.staged_forwarding();
+        Arc::clone(&self.settings).read_owned().await
+    }
+
+    pub async fn write_owned(self: Arc<Self>) -> tokio::sync::OwnedRwLockWriteGuard<AppSettings> {
+        self.staged_forwarding();
+        Arc::clone(&self.settings).write_owned().await
+    }
+
+    pub fn try_read_owned(
+        self: Arc<Self>,
+    ) -> Result<tokio::sync::OwnedRwLockReadGuard<AppSettings>, tokio::sync::TryLockError> {
+        self.staged_forwarding();
+        Arc::clone(&self.settings).try_read_owned()
+    }
+}
+
+pub type SettingsState = Arc<SettingsOwner>;
 
 pub fn new_settings_state(settings: AppSettings) -> SettingsState {
-    Arc::new(RwLock::new(settings))
+    Arc::new(SettingsOwner::staged(settings))
 }
 
 #[cfg(test)]
@@ -15691,9 +16902,13 @@ mod tests {
         let fixture = tempfile::tempdir().unwrap();
         let path = fixture.path().join("settings.json");
         let mut debug = Some(debug);
-        let file =
-            super::SettingsFileLock::acquire_file(&path, std::time::Duration::ZERO, &mut debug)
-                .unwrap_or_else(|_| panic!("disconnected diagnostics affected acquisition"));
+        let file = super::SettingsFileLock::acquire_file(
+            &path,
+            super::SETTINGS_LOCK_FILE_NAME,
+            std::time::Duration::ZERO,
+            &mut debug,
+        )
+        .unwrap_or_else(|_| panic!("disconnected diagnostics affected acquisition"));
         drop(super::SettingsFileLock {
             file: Some(file),
             debug,
@@ -33865,3 +35080,512 @@ mod p02_duplicate_row_tests {
         assert!(retain(&mut ordinary, &raw).is_err());
     }
 }
+
+#[cfg(test)]
+mod e2992_p41_owner_tests {
+    use super::*;
+    use serde_json::json;
+    use std::time::Duration;
+
+    const WAIT: Duration = Duration::from_secs(5);
+    const SOFT_DEADLINE: Duration = Duration::from_secs(30);
+
+    fn instance(base: Option<Value>) -> (tempfile::TempDir, PathBuf) {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join(SETTINGS_FILE_NAME);
+        if let Some(base) = base {
+            write_json(&path, &base);
+        }
+        (dir, path)
+    }
+
+    fn write_json(path: &Path, value: &Value) {
+        let bytes = serde_json::to_vec_pretty(value).unwrap();
+        std::fs::write(path, bytes).unwrap();
+    }
+
+    fn write_raw(path: &Path, bytes: &[u8]) -> Result<(), String> {
+        std::fs::write(path, bytes).map_err(|error| error.to_string())
+    }
+
+    fn voice_enabled(owner: &SettingsOwner) -> bool {
+        let snapshot = owner.blocking_current_snapshot().unwrap();
+        snapshot.voice_to_text_enabled
+    }
+
+    /// A complete, valid base: `defaultShell` and `defaultShellArgs` are required
+    /// fields, so a partial object is a present-but-undecodable base the owner
+    /// must refuse, never a fixture.
+    fn voice(enabled: bool) -> Value {
+        let mut base = serde_json::to_value(AppSettings::default()).unwrap();
+        base["voiceToTextEnabled"] = json!(enabled);
+        base
+    }
+
+    fn counts(owner: &SettingsOwner) -> [u64; 3] {
+        let counters = &owner.counters;
+        [
+            counters.transaction_lock_attempts.load(Ordering::Relaxed),
+            counters.captures.load(Ordering::Relaxed),
+            counters.source_reads.load(Ordering::Relaxed),
+        ]
+    }
+
+    fn overlay_path(path: &Path) -> PathBuf {
+        path.with_file_name(SETTINGS_LOCAL_OVERRIDE_FILE_NAME)
+    }
+
+    /// Link-free physical aliases of `dir`: its canonical spelling when that
+    /// differs textually (Windows verbatim prefix, macOS `/var` -> `/private/var`),
+    /// and a case variant of its leaf when the filesystem resolves it to the same
+    /// directory. Each is checked to canonicalize to the same directory.
+    fn physical_aliases(dir: &Path) -> Vec<PathBuf> {
+        let canonical = std::fs::canonicalize(dir).unwrap();
+        let mut aliases = Vec::new();
+        if canonical != dir {
+            aliases.push(canonical.clone());
+        }
+        let leaf = dir.file_name().unwrap().to_string_lossy().into_owned();
+        let upper = leaf.to_ascii_uppercase();
+        let flipped = if upper != leaf {
+            upper
+        } else {
+            leaf.to_ascii_lowercase()
+        };
+        let variant = dir.with_file_name(flipped);
+        let same = std::fs::canonicalize(&variant).is_ok_and(|resolved| resolved == canonical);
+        if variant != dir && same {
+            aliases.push(variant);
+        }
+        aliases
+    }
+
+    fn is_compose(error: &SettingsOwnerError) -> bool {
+        matches!(error, SettingsOwnerError::Compose(_))
+    }
+
+    #[test]
+    fn e2992_p41_capture_locked_once() {
+        // Positive control: only the base exists; the other three sources are
+        // legitimately absent and compose without defaults replacing the base.
+        let (_dir, path) = instance(Some(voice(true)));
+        let owner = SettingsOwner::open_active(&path, WAIT).unwrap();
+        assert_eq!(counts(&owner), [1, 1, 4]);
+        let snapshot = owner.blocking_current_snapshot().unwrap();
+        assert!(snapshot.voice_to_text_enabled);
+        assert!(snapshot.project_paths.is_empty());
+        assert_eq!(snapshot.agents_layer, AgentsLayerState::Readable);
+
+        // One operation: one transactional lock attempt and one capture; the
+        // locked halves consume the guard and never reacquire the lock.
+        let report = owner
+            .with_operation_blocking("probe", |current, guard| {
+                assert!(current.voice_to_text_enabled);
+                let reacquire = SettingsFileLock::acquire(guard.settings_path(), Duration::ZERO);
+                assert!(reacquire.is_err(), "the guard holds the transactional lock");
+                Ok(())
+            })
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::Committed);
+        assert_eq!(counts(&owner), [2, 2, 8]);
+
+        // A corrupt present agents file is not absence: Recovery refuses it and
+        // leaves its bytes untouched; Startup keeps it as metadata.
+        let agents = agents_instance_path(&path);
+        std::fs::write(&agents, b"{not json").unwrap();
+        let error = owner
+            .with_operation_blocking("corrupt-agents", |_, _| Ok(()))
+            .unwrap_err();
+        assert!(is_compose(&error), "{error:?}");
+        assert_eq!(std::fs::read(&agents).unwrap(), b"{not json");
+        assert!(voice_enabled(&owner));
+        owner.drain_blocking(SOFT_DEADLINE);
+        let startup = SettingsOwner::open_active(&path, WAIT).unwrap();
+        let snapshot = startup.blocking_current_snapshot().unwrap();
+        assert!(matches!(
+            snapshot.agents_layer,
+            AgentsLayerState::Unreadable { .. }
+        ));
+        startup.drain_blocking(SOFT_DEADLINE);
+        std::fs::remove_file(agents).unwrap();
+
+        // A malformed overlay is a Startup warning and a Recovery failure.
+        std::fs::write(overlay_path(&path), b"[1, 2").unwrap();
+        let startup = SettingsOwner::open_active(&path, WAIT).unwrap();
+        let snapshot = startup.blocking_current_snapshot().unwrap();
+        assert!(snapshot.local_overlay_state.rejection().is_some());
+        let error = startup
+            .with_operation_blocking("malformed-overlay", |_, _| Ok(()))
+            .unwrap_err();
+        assert!(is_compose(&error), "{error:?}");
+        startup.drain_blocking(SOFT_DEADLINE);
+        std::fs::remove_file(overlay_path(&path)).unwrap();
+
+        // A corrupt base never becomes defaults.
+        let corrupt: &[u8] = b"{\"voiceToTextEnabled\": tru";
+        std::fs::write(&path, corrupt).unwrap();
+        let error = SettingsOwner::open_active(&path, WAIT).unwrap_err();
+        assert!(is_compose(&error), "{error:?}");
+        assert_eq!(std::fs::read(&path).unwrap(), corrupt);
+
+        // A source replaced by a non-regular object is a capture error.
+        std::fs::remove_file(&path).unwrap();
+        std::fs::create_dir(&path).unwrap();
+        let error = SettingsOwner::open_active(&path, WAIT).unwrap_err();
+        assert!(matches!(error, SettingsOwnerError::Capture(_)), "{error:?}");
+    }
+
+    #[test]
+    fn e2992_p41_publisher_recovery() {
+        let publishers = [
+            SettingsPublisher::PreflightBackup,
+            SettingsPublisher::ProjectPathsMigration,
+            SettingsPublisher::ProjectPathsBackup,
+            SettingsPublisher::ProjectPathsCleanup,
+            SettingsPublisher::BlockingMenusExport,
+            SettingsPublisher::AgentsExport,
+            SettingsPublisher::AgentsSave,
+            SettingsPublisher::SettingsSave,
+            SettingsPublisher::AuxCache,
+            SettingsPublisher::Stamp,
+        ];
+        for publisher in publishers {
+            let (_dir, path) = instance(Some(voice(true)));
+            let owner = publisher_partial_cases(&path, publisher);
+            publisher_invalid_recapture(&owner, &path, publisher);
+        }
+        panic_classification_cases();
+    }
+
+    /// Rejected before any publisher, then PartialOrUnknown after `publisher`
+    /// wrote and the principal failed or panicked.
+    fn publisher_partial_cases(path: &Path, publisher: SettingsPublisher) -> Arc<SettingsOwner> {
+        let owner = SettingsOwner::open_active(path, WAIT).unwrap();
+        let epoch = owner.identity().epoch;
+
+        // Failure before any publisher: Rejected, RAM and epoch unchanged.
+        let report = owner
+            .with_operation_blocking("pre", |_, _| -> Result<(), String> {
+                Err("validation".to_string())
+            })
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::Rejected);
+        assert_eq!(owner.identity().epoch, epoch);
+
+        // A publisher wrote, then the principal TempWrite failed. The owner
+        // keeps its authority through recovery and adopts reality as
+        // PartialOrUnknown, never as success or rollback.
+        let alias = path.to_path_buf();
+        let report = owner
+            .with_operation_blocking("partial", move |_, guard| -> Result<(), String> {
+                guard.publish(publisher, |settings_path| {
+                    write_json(settings_path, &voice(false));
+                    let second = SettingsOwner::open_active(&alias, Duration::ZERO);
+                    let held = matches!(second, Err(SettingsOwnerError::LifecycleTimedOut));
+                    assert!(held, "authority is retained during the operation");
+                    Ok(())
+                })?;
+                assert_eq!(guard.effect(), OperationEffect::Possible);
+                Err("TempWrite".to_string())
+            })
+            .unwrap();
+        assert_eq!(
+            report.outcome,
+            OperationOutcome::PartialOrUnknown,
+            "{publisher:?}"
+        );
+        assert_eq!(report.availability, SettingsAvailability::Ready);
+        assert!(report.error.is_some());
+        assert!(!voice_enabled(&owner));
+
+        // A panic after a publisher is classified, never abandoned.
+        let report = owner
+            .with_operation_blocking("panic", move |_, guard| -> Result<(), String> {
+                guard.publish(publisher, |settings_path| {
+                    write_json(settings_path, &voice(true));
+                    Ok(())
+                })?;
+                panic!("injected");
+            })
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::PartialOrUnknown);
+        assert!(voice_enabled(&owner));
+        owner
+    }
+
+    /// An invalid recapture after `publisher` is sticky Invalid until close.
+    fn publisher_invalid_recapture(
+        owner: &SettingsOwner,
+        path: &Path,
+        publisher: SettingsPublisher,
+    ) {
+        // An invalid recapture is Invalid: no current read, no default save.
+        let report = owner
+            .with_operation_blocking("corrupt", move |_, guard| -> Result<(), String> {
+                guard.publish(publisher, |settings_path| write_raw(settings_path, b"{"))?;
+                Err("ReDecode".to_string())
+            })
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::PartialOrUnknown);
+        assert_eq!(report.availability, SettingsAvailability::Invalid);
+        assert_eq!(
+            owner.blocking_current_snapshot().unwrap_err(),
+            SettingsOwnerError::Unavailable(SettingsAvailability::Invalid)
+        );
+        assert_eq!(std::fs::read(path).unwrap(), b"{");
+        let denied = owner.with_operation_blocking("after", |_, _| Ok(()));
+        assert!(matches!(
+            denied,
+            Err(SettingsOwnerError::Unavailable(
+                SettingsAvailability::Invalid
+            ))
+        ));
+        // The authority is still held until the owner closes.
+        let second = SettingsOwner::open_active(path, Duration::ZERO);
+        assert!(matches!(second, Err(SettingsOwnerError::LifecycleTimedOut)));
+        owner.drain_blocking(SOFT_DEADLINE);
+        assert_eq!(owner.availability(), SettingsAvailability::Closed);
+    }
+
+    /// The async form, then panics in the admission capture, in the recapture
+    /// and before the guards.
+    fn panic_classification_cases() {
+        // The async form runs the same body on a blocking worker.
+        let (_dir, path) = instance(Some(voice(false)));
+        let owner = SettingsOwner::open_active(&path, WAIT).unwrap();
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let operation = owner.with_operation("async", |_, guard| {
+            guard.publish(SettingsPublisher::SettingsSave, |settings_path| {
+                write_json(settings_path, &voice(true));
+                Ok(())
+            })
+        });
+        let report = runtime.block_on(operation).unwrap();
+        assert_eq!(report.outcome, OperationOutcome::Committed);
+        let snapshot = runtime.block_on(owner.current_snapshot()).unwrap();
+        assert!(snapshot.voice_to_text_enabled);
+
+        // A panic in the admission capture, outside the builder and before any
+        // publisher, is classified under the guards: Rejected, RAM kept, Ready.
+        OWNER_CAPTURE_PANIC.with(|armed| armed.set(true));
+        let report = owner
+            .with_operation_blocking("capture-panic", |_, _| Ok(()))
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::Rejected);
+        assert_eq!(report.availability, SettingsAvailability::Ready);
+        assert!(report.error.is_some());
+        assert!(voice_enabled(&owner));
+
+        // A panic in the recapture after a publisher cannot adopt reality:
+        // PartialOrUnknown and sticky Invalid, never success.
+        let report = owner
+            .with_operation_blocking("recapture-panic", |_, guard| {
+                guard.publish(SettingsPublisher::SettingsSave, |settings_path| {
+                    write_json(settings_path, &voice(false));
+                    Ok(())
+                })?;
+                OWNER_CAPTURE_PANIC.with(|armed| armed.set(true));
+                Ok(())
+            })
+            .unwrap();
+        assert_eq!(report.outcome, OperationOutcome::PartialOrUnknown);
+        assert_eq!(report.availability, SettingsAvailability::Invalid);
+        assert!(report.value.is_none());
+        assert!(owner.blocking_current_snapshot().is_err());
+        // Every admission was released, so the drain terminates.
+        owner.drain_blocking(SOFT_DEADLINE);
+        assert_eq!(owner.availability(), SettingsAvailability::Closed);
+
+        // A panic before the guards (a blocking state guard requested from
+        // inside a runtime) published nothing: a worker error, Ready kept, and
+        // the drain still terminates.
+        let (_dir, path) = instance(Some(voice(true)));
+        let owner = SettingsOwner::open_active(&path, WAIT).unwrap();
+        let in_runtime =
+            runtime.block_on(async { owner.with_operation_blocking("in-runtime", |_, _| Ok(())) });
+        assert!(
+            matches!(in_runtime, Err(SettingsOwnerError::Worker(_))),
+            "{in_runtime:?}"
+        );
+        assert_eq!(owner.availability(), SettingsAvailability::Ready);
+        assert!(voice_enabled(&owner));
+        owner.drain_blocking(SOFT_DEADLINE);
+    }
+
+    #[test]
+    fn e2992_p41_memory_zero_io() {
+        let (dir, path) = instance(Some(voice(true)));
+        let owner = SettingsOwner::open_active(&path, WAIT).unwrap();
+        // The owner's transactions lock the physical spelling (macOS temp dirs
+        // sit under the `/var` link, which the transactional lock rejects).
+        let physical = std::fs::canonicalize(dir.path())
+            .unwrap()
+            .join(SETTINGS_FILE_NAME);
+        let held = SettingsFileLock::acquire(&physical, Duration::from_secs(2)).unwrap();
+        let before = counts(&owner);
+        for _ in 0..100 {
+            assert!(voice_enabled(&owner));
+            let retention = owner.blocking_current_retention().unwrap();
+            assert_eq!(retention.source_error, None);
+            let permit = owner.try_current_read_owned().unwrap();
+            assert!(permit.voice_to_text_enabled);
+            assert!(Arc::ptr_eq(permit.owner(), &owner));
+            drop(permit);
+            assert!(owner.blocking_current_read().is_ok());
+            assert!(owner.try_current_read().is_ok());
+        }
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        runtime.block_on(async {
+            for _ in 0..100 {
+                assert!(owner.current_snapshot().await.is_ok());
+                assert!(owner.current_retention().await.is_ok());
+            }
+        });
+        assert_eq!(
+            counts(&owner),
+            before,
+            "memory reads touch no source or lock"
+        );
+
+        // Positive control: a mutation does attempt the transactional lock.
+        let error = owner
+            .with_operation_blocking("apply", |_, _| Ok(()))
+            .unwrap_err();
+        assert!(
+            matches!(error, SettingsOwnerError::TransactionLock(_)),
+            "{error:?}"
+        );
+        assert_eq!(counts(&owner), [before[0] + 1, before[1], before[2]]);
+        drop(held);
+
+        // Invalid denies every read without touching the filesystem.
+        let report = owner
+            .with_operation_blocking("corrupt", |_, guard| {
+                guard.publish(SettingsPublisher::SettingsSave, |settings_path| {
+                    write_raw(settings_path, b"[")
+                })
+            })
+            .unwrap();
+        assert_eq!(report.availability, SettingsAvailability::Invalid);
+        let before = counts(&owner);
+        assert!(owner.blocking_current_snapshot().is_err());
+        assert!(owner.blocking_current_retention().is_err());
+        assert!(owner.try_current_read_owned().is_err());
+        assert!(owner.try_current_read().is_err());
+        assert_eq!(counts(&owner), before);
+        owner.drain_blocking(SOFT_DEADLINE);
+
+        // StagedCompatibility: the P40 constructor keeps the forwarding surface
+        // and refuses managed operations before any lock attempt.
+        let staged = new_settings_state(AppSettings::default());
+        assert!(!staged.identity().managed_operations_enabled);
+        assert_eq!(staged.availability(), SettingsAvailability::Ready);
+        staged.blocking_write().voice_to_text_enabled = true;
+        assert!(staged.blocking_read().voice_to_text_enabled);
+        assert!(voice_enabled(&staged));
+        let refused = staged.with_operation_blocking("staged", |_, _| Ok(()));
+        assert!(matches!(
+            refused,
+            Err(SettingsOwnerError::OperationsNotEnabled)
+        ));
+        assert_eq!(counts(&staged), [0, 0, 0]);
+    }
+
+    #[test]
+    fn e2992_p41_lifecycle_alias() {
+        let (dir, path) = instance(Some(voice(true)));
+        std::fs::create_dir(dir.path().join("sub")).unwrap();
+        let first = SettingsOwner::open_active(&path, WAIT).unwrap();
+        let first_id = first.identity().instance_id;
+        let sidecar = dir.path().join(SETTINGS_OWNER_LOCK_FILE_NAME);
+        assert!(sidecar.is_file());
+        let base = std::fs::read(&path).unwrap();
+        let before = counts(&first);
+
+        // Physical aliases: different spellings that name the same directory
+        // object. Each one meets the same authority, times out, and reads no
+        // source.
+        #[allow(unused_mut)]
+        let mut aliases = physical_aliases(dir.path());
+        if cfg!(windows) {
+            // NTFS is case-insensitive and canonicalize yields a verbatim
+            // spelling, so Windows always has both aliases.
+            assert_eq!(aliases.len(), 2, "{aliases:?}");
+        }
+        // Linux temp directories are usually case-sensitive and link-free, so
+        // the test owns one directory symlink to the instance directory. It
+        // lives in a separate test-owned tempdir (no cycle inside the target)
+        // and is the waiter's alias even when an ambient alias also exists.
+        #[cfg(target_os = "linux")]
+        let (link_dir, link) = {
+            let link_dir = tempfile::tempdir().unwrap();
+            let link = link_dir.path().join("owner-alias");
+            let target = std::fs::canonicalize(dir.path()).unwrap();
+            std::os::unix::fs::symlink(&target, &link).unwrap();
+            assert!(std::fs::symlink_metadata(&link)
+                .unwrap()
+                .file_type()
+                .is_symlink());
+            assert_eq!(std::fs::canonicalize(&link).unwrap(), target);
+            assert_ne!(link, dir.path());
+            aliases.insert(0, link.clone());
+            (link_dir, link)
+        };
+        assert!(!aliases.is_empty(), "every platform needs a physical alias");
+        for alias_dir in &aliases {
+            let alias = alias_dir.join(SETTINGS_FILE_NAME);
+            let error = SettingsOwner::open_active(&alias, Duration::from_millis(100));
+            assert_eq!(error.unwrap_err(), SettingsOwnerError::LifecycleTimedOut);
+        }
+        let alias = aliases[0].join(SETTINGS_FILE_NAME);
+
+        // An identity that cannot be established is an error, never "free".
+        let parent_alias = dir.path().join("sub").join("..").join(SETTINGS_FILE_NAME);
+        let error = SettingsOwner::open_active(&parent_alias, Duration::ZERO).unwrap_err();
+        assert!(
+            matches!(error, SettingsOwnerError::Lifecycle(_)),
+            "{error:?}"
+        );
+        let missing = dir.path().join("missing").join(SETTINGS_FILE_NAME);
+        let error = SettingsOwner::open_active(&missing, Duration::ZERO).unwrap_err();
+        assert!(
+            matches!(error, SettingsOwnerError::Lifecycle(_)),
+            "{error:?}"
+        );
+
+        // Only the holder touched the sources.
+        assert_eq!(counts(&first), before);
+        assert_eq!(std::fs::read(&path).unwrap(), base);
+
+        // A second handle waits, then acquires once the holder releases.
+        let waiter = std::thread::spawn(move || SettingsOwner::open_active(&alias, WAIT));
+        std::thread::sleep(Duration::from_millis(200));
+        assert!(!waiter.is_finished());
+        first.drain_blocking(SOFT_DEADLINE);
+        let second = waiter.join().unwrap().unwrap();
+        assert_eq!(second.availability(), SettingsAvailability::Ready);
+        assert_ne!(second.identity().instance_id, first_id);
+        assert!(voice_enabled(&second));
+        second.drain_blocking(SOFT_DEADLINE);
+        // The stable sidecar is never unlinked.
+        assert!(sidecar.is_file());
+
+        // Remove only the link, never its target, then close its tempdir.
+        #[cfg(target_os = "linux")]
+        {
+            std::fs::remove_file(&link).unwrap();
+            let gone = std::fs::symlink_metadata(&link).unwrap_err();
+            assert_eq!(gone.kind(), std::io::ErrorKind::NotFound);
+            assert!(dir.path().is_dir());
+            assert_eq!(std::fs::read(&path).unwrap(), base);
+            assert!(sidecar.is_file());
+            link_dir.close().unwrap();
+        }
+    }
+}
diff --git a/src/sidebar/components/ProjectPanel.context-menu.test.tsx b/src/sidebar/components/ProjectPanel.context-menu.test.tsx
index 8f9465bd5..9da44f167 100644
--- a/src/sidebar/components/ProjectPanel.context-menu.test.tsx
+++ b/src/sidebar/components/ProjectPanel.context-menu.test.tsx
@@ -2030,10 +2030,25 @@ describe("ProjectPanel replica context menu — session actions (#1673)", () =>
     let addSpy: ReturnType<typeof vi.spyOn> | null = null;
     let removeSpy: ReturnType<typeof vi.spyOn> | null = null;
     let rafSpy: ReturnType<typeof vi.spyOn> | null = null;
+    let cancelRafSpy: ReturnType<typeof vi.spyOn> | null = null;
+    let nextFrameId = 0;
+    const frameRequests: { id: number; callback: FrameRequestCallback }[] = [];
+    const pendingFrames = new Map<number, FrameRequestCallback>();
+    const cancelledFrames = new Set<number>();
     let replacementRectSpy: ReturnType<typeof vi.spyOn> | null = null;
     try {
       Object.defineProperty(window, "innerWidth", { configurable: true, value: 300 });
       Object.defineProperty(window, "innerHeight", { configurable: true, value: 200 });
+      rafSpy = vi.spyOn(window, "requestAnimationFrame").mockImplementation((callback) => {
+        const id = ++nextFrameId;
+        frameRequests.push({ id, callback });
+        pendingFrames.set(id, callback);
+        return id;
+      });
+      cancelRafSpy = vi.spyOn(window, "cancelAnimationFrame").mockImplementation((id) => {
+        cancelledFrames.add(id);
+        pendingFrames.delete(id);
+      });
       const fake = await setupPanel();
       fake.onInvoke("get_settings", () => gate.promise);
       addSpy = vi.spyOn(window, "addEventListener");
@@ -2041,18 +2056,15 @@ describe("ProjectPanel replica context menu — session actions (#1673)", () =>
 
       const menuA = await openMenu(undefined, 40, 120);
       const menuAListeners = await captureDismissListeners(addSpy);
-      const rafCallbacks: FrameRequestCallback[] = [];
-      rafSpy = vi.spyOn(window, "requestAnimationFrame").mockImplementation((callback) => {
-        rafCallbacks.push(callback);
-        return rafCallbacks.length;
-      });
       click(findExactMenuButton(menuA, "Attach Telegram")!);
       await waitFor(() => expect(fake.callsFor("get_settings")).toHaveLength(1));
+      const responseOffset = frameRequests.length;
       settled = true;
       gate.resolve({ telegramBots: [botOne, botTwo] });
       await waitFor(() => expect(menuA.querySelectorAll(".settings-color-dot")).toHaveLength(2));
-      expect(rafCallbacks).toHaveLength(1);
-      const staleReclamp = rafCallbacks.shift()!;
+      const responseRequests = frameRequests.slice(responseOffset);
+      expect(responseRequests).toHaveLength(1);
+      const staleReclamp = responseRequests[0];
 
       dispatchDismiss("escape");
       await waitFor(() => expect(replicaMenu()).toBeNull());
@@ -2080,7 +2092,9 @@ describe("ProjectPanel replica context menu — session actions (#1673)", () =>
         toJSON: () => ({}),
       } as DOMRect);
 
-      staleReclamp(0);
+      // Deliberately deliver the old response callback even if dismissal cancelled it.
+      pendingFrames.delete(staleReclamp.id);
+      staleReclamp.callback(0);
       await Promise.resolve();
       expect(replicaMenu()).toBe(replacement);
       expect(replacement.style.left).toBe(left);
@@ -2097,15 +2111,26 @@ describe("ProjectPanel replica context menu — session actions (#1673)", () =>
       await waitFor(() => expect(replicaMenu()).toBeNull());
       expect(listenerWasRemoved(removeSpy, "click", replacementListeners.click)).toBe(true);
     } finally {
-      if (!settled) gate.resolve({ telegramBots: [] });
-      if (replicaMenu()) dispatchDismiss("click");
-      replacementRectSpy?.mockRestore();
-      rafSpy?.mockRestore();
-      removeSpy?.mockRestore();
-      addSpy?.mockRestore();
-      if (innerWidthDescriptor) Object.defineProperty(window, "innerWidth", innerWidthDescriptor);
-      if (innerHeightDescriptor) {
-        Object.defineProperty(window, "innerHeight", innerHeightDescriptor);
+      try {
+        if (!settled) gate.resolve({ telegramBots: [] });
+        if (replicaMenu()) dispatchDismiss("click");
+      } finally {
+        try {
+          rendered?.cleanup();
+          rendered = null;
+        } finally {
+          pendingFrames.clear();
+          cancelledFrames.clear();
+          replacementRectSpy?.mockRestore();
+          cancelRafSpy?.mockRestore();
+          rafSpy?.mockRestore();
+          removeSpy?.mockRestore();
+          addSpy?.mockRestore();
+          if (innerWidthDescriptor) Object.defineProperty(window, "innerWidth", innerWidthDescriptor);
+          if (innerHeightDescriptor) {
+            Object.defineProperty(window, "innerHeight", innerHeightDescriptor);
+          }
+        }
       }
     }
   });