diff --git a/control/crates/layerfs-control/tests/kernel_application.rs b/control/crates/layerfs-control/tests/kernel_application.rs
index 0e06f0ca9c..84077017d4 100644
--- a/control/crates/layerfs-control/tests/kernel_application.rs
+++ b/control/crates/layerfs-control/tests/kernel_application.rs
@@ -56,6 +56,7 @@ fn actual_daemon_install_hello_bind_status_no_constructor_and_session_end() {
             connections: 2,
             handshake_ms: 3000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 64 * 1024,
             namespaces: 8,
             pager_kib: 1024,
diff --git a/control/crates/layerfs-control/tests/observed_application.rs b/control/crates/layerfs-control/tests/observed_application.rs
index 8ba3599993..caf962a20f 100644
--- a/control/crates/layerfs-control/tests/observed_application.rs
+++ b/control/crates/layerfs-control/tests/observed_application.rs
@@ -78,6 +78,7 @@ fn configure(f: &support::Fixture, deployment: &Deployment) -> PathBuf {
             connections: 4,
             handshake_ms: 3000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 64 * 1024,
             namespaces: 8,
             pager_kib: 1024,
diff --git a/control/crates/layerfs-control/tests/provider_ownership.rs b/control/crates/layerfs-control/tests/provider_ownership.rs
index 457a9aa9fc..5f83b17d17 100644
--- a/control/crates/layerfs-control/tests/provider_ownership.rs
+++ b/control/crates/layerfs-control/tests/provider_ownership.rs
@@ -109,6 +109,7 @@ fn native_create(
             connections: 2,
             handshake_ms: 1000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 65536,
             namespaces: 2,
             pager_kib: 1024,
diff --git a/control/crates/layerfs-control/tests/r5_1_actual_daemon_commit.rs b/control/crates/layerfs-control/tests/r5_1_actual_daemon_commit.rs
index 29d709b27c..30846e6291 100644
--- a/control/crates/layerfs-control/tests/r5_1_actual_daemon_commit.rs
+++ b/control/crates/layerfs-control/tests/r5_1_actual_daemon_commit.rs
@@ -118,6 +118,7 @@ fn start(f: &support::Fixture, mounts: &Path) -> (Process, SocketAddr) {
             connections: 4,
             handshake_ms: 3000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 64 * 1024,
             namespaces: 8,
             pager_kib: 1024,
diff --git a/control/crates/layerfs-sandbox/src/daemon/operator.rs b/control/crates/layerfs-sandbox/src/daemon/operator.rs
index 36f8ee74cd..d706535959 100644
--- a/control/crates/layerfs-sandbox/src/daemon/operator.rs
+++ b/control/crates/layerfs-sandbox/src/daemon/operator.rs
@@ -3,9 +3,10 @@ use crate::{internal::Control, SdkError, SdkResult};
 use layerfs_bridge::control::{SdkRefusal, SdkValue};
 
 /// Calls one daemon backend method: the read model that lists its
-/// Workspaces (`sandbox.list_workspaces`), or the operator view of control
+/// Workspaces (`sandbox.list_workspaces`), the operator view of control
 /// admission and retained connection failures
-/// (`backend.daemon.connection_failures`, #540). Any other method is refused
+/// (`backend.daemon.connection_failures`, #540), or the operator reclaim
+/// (`backend.daemon.reclaim`, #547 slice 5). Any other method is refused
 /// before it is sent.
 pub fn call(control: &mut Control, method: &str, args: SdkValue) -> SdkResult {
     match method {
@@ -14,6 +15,11 @@ pub fn call(control: &mut Control, method: &str, args: SdkValue) -> SdkResult {
         layerfs_bridge::control::OPERATION => {
             crate::daemon::call::exchange_call(control, method, args)
         }
+        // One generation-checked, one-shot reclaim of a named worker (#547
+        // slice 5): never restarts the daemon and never touches Workspaces.
+        layerfs_bridge::control::RECLAIM => {
+            crate::daemon::call::exchange_call(control, method, args)
+        }
         "sandbox.list_workspaces" => crate::daemon::call::exchange_call(control, method, args),
         _ => Err(SdkError::Refused(SdkRefusal {
             code: "UnknownMethod".into(),
diff --git a/control/crates/layerfs-sandbox/src/daemon/pool.rs b/control/crates/layerfs-sandbox/src/daemon/pool.rs
index ad03cb7001..931ea78828 100644
--- a/control/crates/layerfs-sandbox/src/daemon/pool.rs
+++ b/control/crates/layerfs-sandbox/src/daemon/pool.rs
@@ -51,61 +51,53 @@ pub const ADMISSION_WAIT: Duration = Duration::from_secs(60);
 /// before it ends, and its `Busy` refusal is not retried.
 pub const LEND_WAIT: Duration = ADMISSION_WAIT;
 
-/// Why an open ended before the daemon's Hello reply. Every one means the
-/// call was not admitted and nothing was attempted, so opening again is
-/// safe; they name different observable causes (#540).
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-enum NotAdmitted {
-    /// The daemon had no vacant pool slot. `retained` is exact when the
-    /// advisory capacity notice arrived, and unknown when only EOF was seen.
-    NoVacantSlot { retained: Option<u8> },
-    /// The connection itself failed before Hello (`BrokenPipe`,
-    /// `ConnectionReset`).
-    ConnectionFailed,
-}
-
-impl NotAdmitted {
-    fn code(self) -> &'static str {
-        match self {
-            Self::NoVacantSlot { .. } => "Capacity",
-            Self::ConnectionFailed => "Unavailable",
-        }
-    }
-
-    fn admission(self) -> &'static str {
-        match self {
-            Self::NoVacantSlot { .. } => "no-vacant-slot",
-            Self::ConnectionFailed => "connection-failed",
-        }
-    }
-}
-
-/// Classify a failed open that never reached the Hello reply. Through a port
-/// proxy the client reads the end (`UnexpectedEof`, a clean close by the
-/// daemon's accept loop); dialing the daemon directly, its handshake write can
-/// meet the closed socket first (`BrokenPipe`, `ConnectionReset`). The old
-/// message claimed "other processes hold all 8 connections"; that is not
-/// knowledge the client has (#540).
-fn not_admitted(refusal: &SdkRefusal) -> Option<(NotAdmitted, &'static str)> {
-    if refusal.code != "Failed" {
-        return None;
-    }
-    if refusal.message.contains("UnexpectedEof") {
-        return Some((
-            NotAdmitted::NoVacantSlot { retained: None },
-            "UnexpectedEof",
-        ));
-    }
-    for kind in ["BrokenPipe", "ConnectionReset"] {
-        if refusal.message.contains(kind) {
-            return Some((NotAdmitted::ConnectionFailed, kind));
+/// The exact refusal for a failed open that never reached the Hello reply,
+/// typed from the channel error rather than from refusal message text (#547
+/// slice 6). Nothing was attempted, so every one is safe to retry; they name
+/// different observable causes.
+fn opening_refusal(error: ChannelError) -> SdkRefusal {
+    match error {
+        // The daemon closed the connection before the handshake completed and
+        // sent no notice: the reason it did not admit this call is unknown, so
+        // the refusal says what was observed, never a guess at "no vacant
+        // slot" (#547 slice 6).
+        ChannelError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof => SdkRefusal {
+            code: "Unknown".to_owned(),
+            message: "the Sandbox's daemon closed the connection before it was admitted; the reason is unknown and the call was not admitted. Retrying is safe"
+                .to_owned(),
+            detail: SdkValue::map([
+                ("observed", SdkValue::text("UnexpectedEof")),
+                ("retryable", SdkValue::Bool(true)),
+                ("retry_after", SdkValue::Int(500)),
+            ]),
+        },
+        // Dialing the daemon directly, its handshake write can meet the closed
+        // socket first: the connection itself failed, not admission.
+        ChannelError::Io(io)
+            if matches!(
+                io.kind(),
+                std::io::ErrorKind::ConnectionReset | std::io::ErrorKind::BrokenPipe
+            ) =>
+        {
+            let kind = io.kind().to_string();
+            SdkRefusal {
+                code: "Unavailable".to_owned(),
+                message: format!(
+                    "the connection to the Sandbox's daemon failed before it was admitted ({kind}); nothing was attempted. Retrying is safe"
+                ),
+                detail: SdkValue::map([
+                    ("observed", SdkValue::text(kind)),
+                    ("retryable", SdkValue::Bool(true)),
+                    ("retry_after", SdkValue::Int(500)),
+                ]),
+            }
         }
+        other => failed(other),
     }
-    None
 }
 
-/// The exact refusal for a decoded advisory capacity notice. This is the
-/// precise form: it names the resource and the retained record count.
+/// The exact refusal for a decoded advisory capacity notice: it names the
+/// resource, the vacant slots and how long to wait (#540, #547 slice 6).
 fn notice_refusal(notice: AdmissionNotice) -> SdkRefusal {
     match notice.reason {
         AdmissionReason::NoPoolSlot => SdkRefusal {
@@ -123,6 +115,12 @@ fn notice_refusal(notice: AdmissionNotice) -> SdkRefusal {
                     SdkValue::Int(i64::from(notice.reserved_vacant)),
                 ),
                 ("retryable", SdkValue::Bool(true)),
+                // An idle pool slot becomes reclaimable after the daemon's
+                // `idle_ttl`, so retry after that, not sooner.
+                (
+                    "retry_after",
+                    SdkValue::Int(i64::from(crate::lifecycle::IDLE_TTL_MS)),
+                ),
                 ("admission", SdkValue::text(notice.reason.as_str())),
             ]),
         },
@@ -134,58 +132,34 @@ fn notice_refusal(notice: AdmissionNotice) -> SdkRefusal {
                 ("resource", SdkValue::Null),
                 ("retained", SdkValue::Null),
                 ("retryable", SdkValue::Bool(true)),
+                ("retry_after", SdkValue::Int(500)),
                 ("admission", SdkValue::text("starting")),
             ]),
         },
-    }
-}
-
-/// The exact refusal after `wait`: what the client observed, not a guess at
-/// what another process is doing.
-fn admission_refusal(wait: Duration, admission: NotAdmitted, kind: &str) -> SdkRefusal {
-    let message = match admission {
-        NotAdmitted::NoVacantSlot { retained } => format!(
-            "the Sandbox's daemon admitted no control connection for {}s: it closed the connection before Hello ({kind}), which it does when it has no vacant control slot{}. The call was not admitted; retrying is safe",
-            wait.as_secs(),
-            retained.map_or(String::new(), |count| format!(
-                " (it reported {count} retained failure record(s))"
-            ))
-        ),
-        NotAdmitted::ConnectionFailed => format!(
-            "the connection to the Sandbox's daemon failed before Hello for {}s ({kind}): the call was not admitted; retrying is safe",
-            wait.as_secs()
-        ),
-    };
-    let retained = match admission {
-        NotAdmitted::NoVacantSlot { retained } => {
-            retained.map_or(SdkValue::Null, |count| SdkValue::Int(i64::from(count)))
-        }
-        _ => SdkValue::Null,
-    };
-    SdkRefusal {
-        code: admission.code().to_owned(),
-        message,
-        detail: SdkValue::map([
-            (
-                "resource",
-                match admission {
-                    NotAdmitted::NoVacantSlot { .. } => SdkValue::text("daemon_control_slot"),
-                    _ => SdkValue::Null,
-                },
+        // A newer daemon's reason this client does not know: report what was
+        // observed, never guess (#547 slice 6 version skew).
+        other => SdkRefusal {
+            code: "Unknown".to_owned(),
+            message: format!(
+                "the Sandbox's daemon refused admission with a reason this client does not know ({:?}); the call was not admitted. Retrying is safe",
+                other
             ),
-            ("retained", retained),
-            ("retryable", SdkValue::Bool(true)),
-            ("admission", SdkValue::text(admission.admission())),
-            ("observed", SdkValue::text(kind)),
-        ]),
+            detail: SdkValue::map([
+                ("observed", SdkValue::text(other.as_str())),
+                ("retryable", SdkValue::Bool(true)),
+                ("retry_after", SdkValue::Int(500)),
+            ]),
+        },
     }
 }
 
 /// Whether this refusal says nothing was attempted and a later open may
 /// succeed. Only these exact refusals are retried.
 fn retryable(refusal: &SdkRefusal) -> bool {
-    matches!(refusal.code.as_str(), "Capacity" | "Unavailable")
-        && refusal.detail.get_bool("retryable")
+    matches!(
+        refusal.code.as_str(),
+        "Capacity" | "Unavailable" | "Unknown"
+    ) && refusal.detail.get_bool("retryable")
 }
 
 impl Dial {
@@ -224,7 +198,7 @@ impl Dial {
         let connection = match initiate_observed(stream, &self.secret, self.peer).result {
             Ok(connection) => connection,
             Err(ChannelError::Admission(notice)) => return Err(notice_refusal(notice)),
-            Err(error) => return Err(failed(error)),
+            Err(error) => return Err(opening_refusal(error)),
         };
         // the waits stay: the caller makes exactly one call and drops it
         let mut control = Control::new(connection);
@@ -249,17 +223,12 @@ impl Dial {
         loop {
             match self.open() {
                 Err(refusal) => {
-                    let legacy = not_admitted(&refusal);
-                    let may_retry = retryable(&refusal) || legacy.is_some();
-                    if may_retry && started.elapsed() < wait {
+                    if retryable(&refusal) && started.elapsed() < wait {
                         std::thread::sleep(pause);
                         pause = (pause * 2).min(Duration::from_millis(500));
                         continue;
                     }
-                    if let Some((admission, kind)) = legacy {
-                        return Err(admission_refusal(wait, admission, kind));
-                    }
-                    // A notice-based or transport refusal is already exact.
+                    // The refusal is already typed and exact (#547 slice 6).
                     return Err(refusal);
                 }
                 other => return other,
@@ -280,7 +249,7 @@ impl Dial {
         let connection = match initiate_observed(stream, &self.secret, self.peer).result {
             Ok(connection) => connection,
             Err(ChannelError::Admission(notice)) => return Err(notice_refusal(notice)),
-            Err(error) => return Err(failed(error)),
+            Err(error) => return Err(opening_refusal(error)),
         };
         connection.clear_io_waits().map_err(failed)?;
         let mut control = Control::new(connection);
diff --git a/control/crates/layerfs-sandbox/src/daemon/values.rs b/control/crates/layerfs-sandbox/src/daemon/values.rs
index fb45b2655c..e0dad01add 100644
--- a/control/crates/layerfs-sandbox/src/daemon/values.rs
+++ b/control/crates/layerfs-sandbox/src/daemon/values.rs
@@ -24,6 +24,23 @@ pub fn refusal(failure: &OperationFailure) -> Option<SdkRefusal> {
             message: refused.detail.clone(),
             detail: SdkValue::map([("phase", SdkValue::text(refused.phase.as_str()))]),
         }),
+        // The request was sent but no reply arrived (the daemon reclaimed or
+        // expired the worker, or it died mid-operation): its effect is
+        // unknown, so the caller observes `Unavailable { outcome: unknown }`
+        // rather than a silent failure (#547 slice 5).
+        OperationCause::Exchange(exchange)
+            if exchange.phase == crate::daemon::internal::ControlPhase::Receive =>
+        {
+            Some(SdkRefusal {
+                code: "Unavailable".to_owned(),
+                message: "the Sandbox's daemon ended this connection before it answered; the operation's outcome is unknown"
+                    .to_owned(),
+                detail: SdkValue::map([
+                    ("outcome", SdkValue::text("unknown")),
+                    ("retryable", SdkValue::Bool(false)),
+                ]),
+            })
+        }
         _ => None,
     }
 }
diff --git a/control/crates/layerfs-sandbox/src/lifecycle.rs b/control/crates/layerfs-sandbox/src/lifecycle.rs
index d986871a5e..30bfba7628 100644
--- a/control/crates/layerfs-sandbox/src/lifecycle.rs
+++ b/control/crates/layerfs-sandbox/src/lifecycle.rs
@@ -49,6 +49,11 @@ pub const DAEMON_CONNECTIONS: u16 = 8;
 /// connection is likely abandoned, short enough that recovery stays bounded.
 pub const IDLE_TTL_MS: u32 = 60_000;
 
+/// A decoded request still in flight this long is expired (#547 slice 5):
+/// long enough that a 64 MiB Commit finishes normally, short enough that a
+/// fence/Store/diagnostic wait cannot hold a pool slot forever.
+pub const REQUEST_DEADLINE_MS: u32 = 300_000;
+
 pub fn setup(
     private: [u8; 32],
     peer: [u8; 32],
@@ -68,6 +73,7 @@ pub fn setup(
             connections: DAEMON_CONNECTIONS,
             handshake_ms: 2000,
             idle_ttl_ms: IDLE_TTL_MS,
+            request_deadline_ms: REQUEST_DEADLINE_MS,
             // The daemon's shared immutable-object cache: a fixed bound, the
             // daemon's own default. 64 KiB held too few directory and inode
             // pages for lookups in a large directory to stay resident
diff --git a/engine/crates/layerfs-bridge/src/control.rs b/engine/crates/layerfs-bridge/src/control.rs
index 3a63b48147..44a2822f4b 100644
--- a/engine/crates/layerfs-bridge/src/control.rs
+++ b/engine/crates/layerfs-bridge/src/control.rs
@@ -165,7 +165,7 @@ fn boolean(value: u8) -> Result<bool, ControlError> {
 }
 
 pub use crate::control_capacity::{
-    AdmissionModel, CapacityModel, RetainedFailureModel, RetainedModel, OPERATION,
+    AdmissionModel, CapacityModel, RetainedFailureModel, RetainedModel, OPERATION, RECLAIM,
 };
 pub use crate::control_forced::{
     AbortDisposition, CommitKnowledge, DetachDisposition, ForceUnmounted, ForcedCleanup,
diff --git a/engine/crates/layerfs-bridge/src/control_capacity.rs b/engine/crates/layerfs-bridge/src/control_capacity.rs
index 61fe9819f3..c777a51507 100644
--- a/engine/crates/layerfs-bridge/src/control_capacity.rs
+++ b/engine/crates/layerfs-bridge/src/control_capacity.rs
@@ -6,10 +6,28 @@
 //! slots by class) and one [`RetainedModel`] (the bounded store of connection
 //! failures that used to retain a slot). It is a plain value object: parse a
 //! daemon answer once, then read typed fields instead of string keys.
+//!
+//! # Version skew
+//!
+//! This model crosses three separately deployed components (the daemon, the
+//! control binding and the doctor), so a field one side adds is read by an
+//! older side as absent:
+//!
+//! | new × old | failure shape |
+//! |---|---|
+//! | new daemon × old control binding | an unknown admission reason reads as `Unknown { observed }`, never a hang; a new `admission`/`retained` field reads as zero |
+//! | old daemon × new control binding | no notice is written before a full pool, so the binding reports `Unknown { observed: UnexpectedEof }`, never "no vacant slot" |
+//! | new control × old worker bundle | a new `outcome`/`retry_after` detail field is absent and reads as null; nothing is retried on the caller's behalf |
+//!
+//! Every field is defaulted when absent so a newer daemon stays legible to an
+//! older reader, and every unknown reason stays a named refusal, never a hang.
 use crate::control::SdkValue;
 
 /// The daemon backend method that returns this read model.
 pub const OPERATION: &str = "backend.daemon.connection_failures";
+/// The daemon backend method that closes one named worker's socket and
+/// records `Reclaimed` (#547 slice 5).
+pub const RECLAIM: &str = "backend.daemon.reclaim";
 
 /// Vacant and in-use control slots by admission class.
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
@@ -26,6 +44,7 @@ pub struct AdmissionModel {
 
 impl AdmissionModel {
     /// The model in `value`, or all zero when the record does not carry one.
+    #[must_use]
     pub fn parse(value: &SdkValue) -> Self {
         let admission = value.get("admission").unwrap_or(&SdkValue::Null);
         Self {
@@ -37,6 +56,7 @@ impl AdmissionModel {
     }
 
     /// The model as the read model's `admission` record.
+    #[must_use]
     pub fn to_sdk(&self) -> SdkValue {
         SdkValue::map([
             ("pool_vacant", SdkValue::Int(self.pool_vacant)),
@@ -47,6 +67,7 @@ impl AdmissionModel {
     }
 
     /// Whether the pool has no vacant slot while it is in use at all.
+    #[must_use]
     pub fn exhausted(&self) -> bool {
         self.pool_vacant == 0 && self.pool_running > 0
     }
@@ -84,6 +105,7 @@ pub struct RetainedFailureModel {
 
 impl RetainedFailureModel {
     /// The record in `value`, or `None` when it is not a failure record.
+    #[must_use]
     pub fn parse(value: &SdkValue) -> Option<Self> {
         Some(Self {
             slot: value.get_int("slot")?,
@@ -102,6 +124,7 @@ impl RetainedFailureModel {
     }
 
     /// The record as the read model's `retained.failures` item.
+    #[must_use]
     pub fn to_sdk(&self) -> SdkValue {
         SdkValue::map([
             ("slot", SdkValue::Int(self.slot)),
@@ -145,6 +168,7 @@ pub struct RetainedModel {
 
 impl RetainedModel {
     /// The model in `value`, or all zero when the record does not carry one.
+    #[must_use]
     pub fn parse(value: &SdkValue) -> Self {
         let retained = value.get("retained").unwrap_or(&SdkValue::Null);
         let failures = match retained.get("failures") {
@@ -165,6 +189,7 @@ impl RetainedModel {
     }
 
     /// The model as the read model's `retained` record.
+    #[must_use]
     pub fn to_sdk(&self) -> SdkValue {
         SdkValue::map([
             ("limit", SdkValue::Int(self.limit)),
@@ -197,6 +222,7 @@ pub struct CapacityModel {
 impl CapacityModel {
     /// Parse one `backend.daemon.connection_failures` answer. A field the
     /// record does not carry reads as zero, so a newer daemon stays legible.
+    #[must_use]
     pub fn parse(value: &SdkValue) -> Self {
         Self {
             admission: AdmissionModel::parse(value),
@@ -205,6 +231,7 @@ impl CapacityModel {
     }
 
     /// The answer a daemon publishes for this model.
+    #[must_use]
     pub fn to_sdk(&self) -> SdkValue {
         SdkValue::map([
             ("admission", self.admission.to_sdk()),
diff --git a/engine/crates/layerfs-bridge/src/daemon_setup.rs b/engine/crates/layerfs-bridge/src/daemon_setup.rs
index 4e0ae16020..746019c790 100644
--- a/engine/crates/layerfs-bridge/src/daemon_setup.rs
+++ b/engine/crates/layerfs-bridge/src/daemon_setup.rs
@@ -6,7 +6,7 @@ use crate::{
     wire::{Reader, Writer},
 };
 use std::fmt;
-const MAGIC: &[u8] = b"LFSD\x08";
+const MAGIC: &[u8] = b"LFSD\x09";
 /// Explicit daemon startup resources; these are windows/admission, not file caps.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct DaemonLimits {
@@ -18,6 +18,10 @@ pub struct DaemonLimits {
     /// A control connection that stays parked in `receive()` (no operation in
     /// flight) this long is reclaimable when admission is refused (#547 slice 4).
     pub idle_ttl_ms: u32,
+    /// A decoded request whose execution is still in flight this long is
+    /// expired: its slot is freed, its socket is closed and the caller
+    /// observes `Unavailable { outcome: unknown }` (#547 slice 5).
+    pub request_deadline_ms: u32,
     /// Shared immutable cache allowance, separate from whole-process memory.
     pub cache_bytes: u64,
     /// Explicit simultaneously admitted Workspace count.
@@ -68,6 +72,7 @@ impl DaemonSetup {
         out.put(&l.connections.to_be_bytes())?;
         out.put(&l.handshake_ms.to_be_bytes())?;
         out.put(&l.idle_ttl_ms.to_be_bytes())?;
+        out.put(&l.request_deadline_ms.to_be_bytes())?;
         out.put(&l.cache_bytes.to_be_bytes())?;
         for v in [l.namespaces, l.pager_kib] {
             out.put(&v.to_be_bytes())?;
@@ -94,6 +99,7 @@ impl DaemonSetup {
             connections: u16::from_be_bytes(input.array()?),
             handshake_ms: u32::from_be_bytes(input.array()?),
             idle_ttl_ms: u32::from_be_bytes(input.array()?),
+            request_deadline_ms: u32::from_be_bytes(input.array()?),
             cache_bytes: u64::from_be_bytes(input.array()?),
             namespaces: u32::from_be_bytes(input.array()?),
             pager_kib: u32::from_be_bytes(input.array()?),
@@ -148,6 +154,7 @@ impl DaemonSetup {
         let l = self.limits;
         if l.handshake_ms == 0
             || l.idle_ttl_ms == 0
+            || l.request_deadline_ms == 0
             || l.connections == 0
             || l.namespaces == 0
             || l.pager_kib == 0
diff --git a/engine/crates/layerfs-bridge/src/duplex/admission.rs b/engine/crates/layerfs-bridge/src/duplex/admission.rs
index ee500342da..89bd36a338 100644
--- a/engine/crates/layerfs-bridge/src/duplex/admission.rs
+++ b/engine/crates/layerfs-bridge/src/duplex/admission.rs
@@ -18,7 +18,11 @@ pub const VERSION: u8 = 1;
 /// Length of the body after the two-byte record header.
 pub const BODY_LEN: usize = 24;
 
-/// Why no connection was admitted.
+/// Why no connection was admitted. A newer daemon may add reasons, so this
+/// enum is non-exhaustive: a client matches it with a wildcard arm and never
+/// constructs it (use [`AdmissionReason::no_pool_slot`] or
+/// [`AdmissionReason::starting`]).
+#[non_exhaustive]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum AdmissionReason {
     /// Every pool slot was busy (or retained; the count is in `retained`).
@@ -28,6 +32,16 @@ pub enum AdmissionReason {
 }
 
 impl AdmissionReason {
+    /// The `NoPoolSlot` reason: the pool is full.
+    pub fn no_pool_slot() -> Self {
+        Self::NoPoolSlot
+    }
+
+    /// The `Starting` reason: the daemon admits no control yet.
+    pub fn starting() -> Self {
+        Self::Starting
+    }
+
     fn of(byte: u8) -> Option<Self> {
         match byte {
             1 => Some(Self::NoPoolSlot),
@@ -62,6 +76,7 @@ pub struct AdmissionNotice {
 
 impl AdmissionNotice {
     /// A notice with the current format version.
+    #[must_use]
     pub fn new(
         reason: AdmissionReason,
         retained: u8,
@@ -79,6 +94,7 @@ impl AdmissionNotice {
 }
 
 /// The exact 24-byte body of `notice`.
+#[must_use]
 pub fn encode(notice: &AdmissionNotice) -> [u8; BODY_LEN] {
     let mut body = [0u8; BODY_LEN];
     body[..16].copy_from_slice(MAGIC);
@@ -92,6 +108,7 @@ pub fn encode(notice: &AdmissionNotice) -> [u8; BODY_LEN] {
 
 /// The notice in `bytes`, or `None` when the record is anything else. A
 /// random record that is not exactly a valid notice is never accepted.
+#[must_use]
 pub fn classify(bytes: &[u8]) -> Option<AdmissionNotice> {
     if bytes.len() != BODY_LEN || &bytes[..16] != MAGIC {
         return None;
diff --git a/engine/crates/layerfs-bridge/tests/admission.rs b/engine/crates/layerfs-bridge/tests/admission.rs
index e4b421da99..59df0ca48d 100644
--- a/engine/crates/layerfs-bridge/tests/admission.rs
+++ b/engine/crates/layerfs-bridge/tests/admission.rs
@@ -6,7 +6,7 @@ use layerfs_bridge::duplex::admission::{
 
 #[test]
 fn encode_and_classify_round_trip() {
-    let notice = AdmissionNotice::new(AdmissionReason::NoPoolSlot, 8, 0, 1);
+    let notice = AdmissionNotice::new(AdmissionReason::no_pool_slot(), 8, 0, 1);
     let body = encode(&notice);
     assert_eq!(body.len(), BODY_LEN);
     assert_eq!(&body[..16], MAGIC);
@@ -17,13 +17,28 @@ fn encode_and_classify_round_trip() {
 fn garbage_is_not_a_notice() {
     assert_eq!(classify(&[0u8; BODY_LEN]), None);
     assert_eq!(classify(&[0u8; 23]), None);
-    let mut body = encode(&AdmissionNotice::new(AdmissionReason::NoPoolSlot, 1, 1, 1));
+    let mut body = encode(&AdmissionNotice::new(
+        AdmissionReason::no_pool_slot(),
+        1,
+        1,
+        1,
+    ));
     body[16] = 2;
     assert_eq!(classify(&body), None);
-    let mut body = encode(&AdmissionNotice::new(AdmissionReason::NoPoolSlot, 1, 1, 1));
+    let mut body = encode(&AdmissionNotice::new(
+        AdmissionReason::no_pool_slot(),
+        1,
+        1,
+        1,
+    ));
     body[17] = 0;
     assert_eq!(classify(&body), None);
-    let mut body = encode(&AdmissionNotice::new(AdmissionReason::NoPoolSlot, 1, 1, 1));
+    let mut body = encode(&AdmissionNotice::new(
+        AdmissionReason::no_pool_slot(),
+        1,
+        1,
+        1,
+    ));
     body[23] = 1;
     assert_eq!(classify(&body), None);
 }
diff --git a/engine/crates/layerfs-bridge/tests/daemon_setup_records.rs b/engine/crates/layerfs-bridge/tests/daemon_setup_records.rs
index c32fe561c3..55aced0beb 100644
--- a/engine/crates/layerfs-bridge/tests/daemon_setup_records.rs
+++ b/engine/crates/layerfs-bridge/tests/daemon_setup_records.rs
@@ -45,6 +45,7 @@ fn setup(store: DaemonStore, route: KernelRoute) -> DaemonSetup {
             connections: 8,
             handshake_ms: 2000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 1 << 20,
             namespaces: 16,
             pager_kib: 1024,
@@ -130,6 +131,6 @@ fn unknown_route_or_store_bytes_and_the_old_magic_are_refused() {
     bad_kind[store_field - 1 + 1] = 9;
     assert!(DaemonSetup::decode(&bad_kind).is_err());
     let mut old = bytes;
-    old[4] = 6;
+    old[4] = 8;
     assert!(DaemonSetup::decode(&old).is_err());
 }
diff --git a/engine/crates/layerfs-daemon/src/application/connection.rs b/engine/crates/layerfs-daemon/src/application/connection.rs
index 9358c54a2d..08050b0fed 100644
--- a/engine/crates/layerfs-daemon/src/application/connection.rs
+++ b/engine/crates/layerfs-daemon/src/application/connection.rs
@@ -3,11 +3,11 @@ use super::{
     owner::{
         disposition, recovery_request, AdmissionClass, Finished, SlotClass, SlotCounts, SlotTicket,
     },
-    retained::RetainedSnapshot,
+    retained::{RetainedFailure, RetainedSnapshot},
     Application, ApplicationError, ConnectionFailure,
 };
 use layerfs_bridge::{
-    control::{Call, DaemonPhase, Reply, Request},
+    control::{Call, DaemonPhase, Reply, Request, RECLAIM},
     duplex::{
         self as transport,
         admission::{self as admission_notice, AdmissionNotice, AdmissionReason},
@@ -150,6 +150,17 @@ impl Application {
             let ticket = {
                 let idle_ttl = Duration::from_millis(u64::from(self.setup.limits.idle_ttl_ms));
                 let mut state = self.state.lock().map_err(|_| ApplicationError::Poisoned)?;
+                // In-flight expiry (#547 slice 5): a pool slot whose decoded
+                // request is still executing past `request_deadline` has an
+                // unknown effect; free it, close its socket and record
+                // `Expired` so the caller observes
+                // `Unavailable { outcome: unknown }` rather than a silent hang.
+                let expired = state.slots.reap_expired(Instant::now());
+                if !expired.is_empty() {
+                    drop(state);
+                    self.expire_in_flight(&expired);
+                    state = self.state.lock().map_err(|_| ApplicationError::Poisoned)?;
+                }
                 let mut picked = state.slots.admit(AdmissionClass::ControlPool);
                 // Idle reclamation (#547 slice 4): a pool slot whose worker
                 // is parked in `receive()` past `idle_ttl` holds nothing in
@@ -159,10 +170,11 @@ impl Application {
                     if !reaped.is_empty() {
                         drop(state);
                         self.close_slot_fences(&reaped);
-                        for &index in &reaped {
+                        for ticket in &reaped {
                             let _ = writeln!(
                                 std::io::stderr().lock(),
-                                "layerfs-daemon: reclaimed idle control slot {index}"
+                                "layerfs-daemon: reclaimed idle control slot {}",
+                                ticket.index
                             );
                         }
                         state = self.state.lock().map_err(|_| ApplicationError::Poisoned)?;
@@ -188,9 +200,9 @@ impl Application {
                         let retained = state.retained.len();
                         state.retained.reap(Instant::now());
                         let reason = if state.startup.phase != DaemonPhase::ControlReady {
-                            AdmissionReason::Starting
+                            AdmissionReason::starting()
                         } else {
-                            AdmissionReason::NoPoolSlot
+                            AdmissionReason::no_pool_slot()
                         };
                         let notice = AdmissionNotice::new(
                             reason,
@@ -206,11 +218,10 @@ impl Application {
                     }
                 }
             };
-            let previous = self
-                .workers
-                .lock()
-                .map_err(|_| ApplicationError::Poisoned)?[ticket.index]
-                .take();
+            let previous = {
+                let mut state = self.state.lock().map_err(|_| ApplicationError::Poisoned)?;
+                state.workers[ticket.index].take()
+            };
             if let Some(worker) = previous {
                 worker.join().map_err(|_| {
                     ApplicationError::Io(std::io::Error::other(
@@ -238,14 +249,15 @@ impl Application {
                         application.connection(stream, ticket)
                     }))
                     .unwrap_or_else(|payload| Err(panic_failure(ticket, payload)));
-                    application.clear_fence(ticket.index);
+                    application.clear_fence(ticket);
                     guard.finish(result);
                 });
             match worker {
                 Ok(worker) => {
-                    self.workers
+                    self.state
                         .lock()
-                        .map_err(|_| ApplicationError::Poisoned)?[ticket.index] = Some(worker)
+                        .map_err(|_| ApplicationError::Poisoned)?
+                        .workers[ticket.index] = Some(worker)
                 }
                 Err(_cause) => {
                     // Spawning dropped the closure, so the guard's `Drop`
@@ -268,35 +280,45 @@ impl Application {
         )
     }
 
-    /// Stores the socket fence for `slot`, so the accept loop can close it to
-    /// reclaim an idle connection (#547 slice 4).
-    fn set_fence(&self, slot: usize, fence: transport::CloseHandle) {
+    /// Stores the socket fence of one worker, keyed by its admission
+    /// generation, so the accept loop can close it to reclaim an idle
+    /// connection or expire an in-flight request (#547 slices 4-5).
+    fn set_fence(&self, ticket: SlotTicket, fence: transport::CloseHandle) {
         if let Ok(mut fences) = self.fences.lock() {
-            if let Some(entry) = fences.get_mut(slot) {
-                *entry = Some(fence);
+            if let Some(entry) = fences.get_mut(ticket.index) {
+                *entry = Some((ticket.generation, fence));
             }
         }
     }
 
-    /// Removes the socket fence for `slot`; its worker is finished with it.
-    fn clear_fence(&self, slot: usize) {
+    /// Removes one worker's socket fence. A stale worker that finished after
+    /// its slot was reclaimed or expired skips a newer worker's fence because
+    /// the generation no longer matches (#547 slice 5).
+    fn clear_fence(&self, ticket: SlotTicket) {
         if let Ok(mut fences) = self.fences.lock() {
-            if let Some(entry) = fences.get_mut(slot) {
-                *entry = None;
+            if let Some(entry) = fences.get_mut(ticket.index) {
+                if matches!(entry, Some((generation, _)) if *generation == ticket.generation) {
+                    *entry = None;
+                }
             }
         }
     }
 
-    /// Closes the socket fence of each named slot, waking its worker parked
-    /// in `receive()`. A missing fence (the worker already ended) is skipped.
-    fn close_slot_fences(&self, slots: &[usize]) {
+    /// Closes the socket fence of each named worker generation, waking one
+    /// parked in `receive()` or a fence wait. A missing or stale fence (the
+    /// worker already ended or its slot was re-admitted) is skipped.
+    fn close_slot_fences(&self, tickets: &[SlotTicket]) {
         let mut fences = self
             .fences
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner());
-        for &index in slots {
-            if let Some(fence) = fences.get_mut(index).and_then(|entry| entry.take()) {
-                let _ = fence.close();
+        for ticket in tickets {
+            if let Some(entry) = fences.get_mut(ticket.index) {
+                if matches!(entry, Some((generation, _)) if *generation == ticket.generation) {
+                    if let Some((_, fence)) = entry.take() {
+                        let _ = fence.close();
+                    }
+                }
             }
         }
     }
@@ -308,11 +330,37 @@ impl Application {
         }
     }
 
-    /// Marks one worker's slot as executing again (not reapable).
-    fn mark_busy(&self, ticket: SlotTicket) {
+    /// Marks one worker's slot as executing one decoded request with its
+    /// expiry deadline (#547 slice 5).
+    fn mark_inflight(&self, ticket: SlotTicket, started: Instant, deadline: Instant) {
         if let Ok(mut state) = self.state.lock() {
-            state.slots.busy(ticket);
+            state.slots.inflight(ticket, started, deadline);
+        }
+    }
+
+    /// Frees every slot whose in-flight request passed `request_deadline`
+    /// (#547 slice 5): closes its socket (waking a worker parked in socket
+    /// I/O; a worker parked elsewhere finishes stale when it returns) and
+    /// records one `Expired` retained failure each. The interrupted
+    /// operation's effect is unknown, so it is never released as clean.
+    fn expire_in_flight(&self, tickets: &[SlotTicket]) {
+        let now = Instant::now();
+        self.close_slot_fences(tickets);
+        let mut state = self
+            .state
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        for ticket in tickets {
+            // Detach the stuck worker: its eventual finish is stale, and the
+            // accept loop must never join a worker a Store/fence wait holds.
+            state.workers[ticket.index] = None;
+            let owner = self.owner_label(*ticket);
+            let recorded = state
+                .retained
+                .record(RetainedFailure::expired(*ticket, &owner), now);
+            let _ = writeln!(std::io::stderr().lock(), "{recorded}");
         }
+        self.changed.notify_all();
     }
 
     fn connection(&self, stream: TcpStream, ticket: SlotTicket) -> Result<(), ConnectionFailure> {
@@ -355,7 +403,7 @@ impl Application {
                 fence_error: None,
                 diagnostic_error: None,
             })?;
-        self.set_fence(slot, fence);
+        self.set_fence(ticket, fence);
         let mut first = true;
         loop {
             // An established connection parked for the next request is
@@ -394,8 +442,13 @@ impl Application {
                     }
                 }
             };
-            // A request arrived: the slot is executing again, not reapable.
-            self.mark_busy(ticket);
+            // A request arrived: the slot is executing one decoded request
+            // under an explicit deadline (in-flight, #547 slice 5), not
+            // reapable as idle.
+            let started = Instant::now();
+            let deadline =
+                started + Duration::from_millis(u64::from(self.setup.limits.request_deadline_ms));
+            self.mark_inflight(ticket, started, deadline);
             if first {
                 if let Err(cause) = settings
                     .set_read_timeout(None)
@@ -625,14 +678,20 @@ impl Application {
                 "reserved recovery slot",
             ));
         }
-        // Operator-visible retained failures live with the application, not
-        // with the Store control Service, so answer this one here.
+        // Operator-visible retained failures and the reclaim verb live with
+        // the application, not with the Store control Service, so answer
+        // these here (#540, #547 slice 5).
         if let Request::Sdk(call) = request {
             if call.method == layerfs_bridge::control::OPERATION {
                 return self
                     .connection_failures()
                     .map(|value| Success::reply(Reply::Sdk(value)));
             }
+            if call.method == RECLAIM {
+                return self
+                    .reclaim(call)
+                    .map(|value| Success::reply(Reply::Sdk(value)));
+            }
         }
         match service {
             Some(service) => service.execute_control(request),
@@ -647,6 +706,79 @@ impl Application {
         }
     }
 
+    /// `backend.daemon.reclaim { slot, generation }` (#547 slice 5): closes
+    /// the named worker's socket, records `Reclaimed`, and frees its slot. It
+    /// is generation-checked (a vacant slot or a newer worker answers
+    /// `Invalid`), one-shot, never restarts the daemon and never touches a
+    /// Workspace.
+    fn reclaim(
+        &self,
+        call: &layerfs_bridge::control::SdkCall,
+    ) -> Result<layerfs_bridge::control::SdkValue, crate::control::Failure> {
+        use crate::control::Failure;
+        use layerfs_bridge::control::{ControlCode, SdkValue};
+        let slot = match call
+            .args
+            .get_int("slot")
+            .and_then(|value| usize::try_from(value).ok())
+        {
+            Some(slot) => slot,
+            None => {
+                return Err(Failure::Rejected(
+                    ControlCode::Invalid,
+                    "reclaim names the slot to close",
+                ))
+            }
+        };
+        let generation = match call
+            .args
+            .get_int("generation")
+            .and_then(|value| u64::try_from(value).ok())
+        {
+            Some(generation) => generation,
+            None => {
+                return Err(Failure::Rejected(
+                    ControlCode::Invalid,
+                    "reclaim names the generation to close",
+                ))
+            }
+        };
+        // Free the named slot and detach its worker under one lock, so the
+        // accept loop can never join a worker this reclaim leaves running.
+        // Then close the socket and record `Reclaimed`.
+        let (ticket, owner) = {
+            let mut state = self.state.lock().map_err(|_| Failure::Poisoned)?;
+            match state.slots.reclaim(slot, generation) {
+                Some(ticket) => {
+                    state.workers[ticket.index] = None;
+                    (ticket, self.owner_label(ticket))
+                }
+                None => {
+                    return Err(Failure::Rejected(
+                        ControlCode::Invalid,
+                        "reclaim generation mismatch",
+                    ))
+                }
+            }
+        };
+        self.close_slot_fences(std::slice::from_ref(&ticket));
+        let now = Instant::now();
+        let recorded = {
+            let mut state = self.state.lock().map_err(|_| Failure::Poisoned)?;
+            let recorded = state
+                .retained
+                .record(RetainedFailure::reclaimed(ticket, &owner), now);
+            self.changed.notify_all();
+            recorded
+        };
+        let _ = writeln!(std::io::stderr().lock(), "{recorded}");
+        Ok(SdkValue::map([
+            ("outcome", SdkValue::text("reclaimed")),
+            ("slot", SdkValue::Int(slot as i64)),
+            ("generation", SdkValue::Int(generation as i64)),
+        ]))
+    }
+
     /// The operator read model of control admission and retained failures.
     fn connection_failures(
         &self,
diff --git a/engine/crates/layerfs-daemon/src/application/owner.rs b/engine/crates/layerfs-daemon/src/application/owner.rs
index cb4ee3c9fe..2e2a97cf9b 100644
--- a/engine/crates/layerfs-daemon/src/application/owner.rs
+++ b/engine/crates/layerfs-daemon/src/application/owner.rs
@@ -69,11 +69,14 @@ impl AdmissionClass {
 /// The lifecycle phase of one admitted (non-vacant) slot.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum Phase {
-    /// Authenticating or executing one request. Nothing here is reapable.
+    /// Authenticating before the first request. Nothing here is reapable.
     Busy,
     /// Blocked in `receive()` awaiting the next request: nothing is in
     /// flight, so the daemon may reclaim it once `idle_ttl` passes (#547).
     Idle { since: Instant },
+    /// Executing one decoded request; past `deadline` its effect is unknown
+    /// and the daemon expires it (#547 slice 5).
+    InFlight { started: Instant, deadline: Instant },
 }
 
 /// One admission slot: free, or running one worker generation in a phase.
@@ -214,12 +217,30 @@ impl Slots {
         running
     }
 
+    /// Marks one running slot as executing one decoded request with its
+    /// expiry deadline (#547 slice 5); a stale ticket changes nothing.
+    pub fn inflight(&mut self, ticket: SlotTicket, started: Instant, deadline: Instant) -> bool {
+        let running = matches!(
+            self.slots.get(ticket.index),
+            Some(SlotState::Running { generation, class, .. })
+                if *generation == ticket.generation && *class == ticket.class
+        );
+        if running {
+            self.slots[ticket.index] = SlotState::Running {
+                generation: ticket.generation,
+                class: ticket.class,
+                phase: Phase::InFlight { started, deadline },
+            };
+        }
+        running
+    }
+
     /// Frees every pool slot whose worker has been parked in `receive()` for
-    /// at least `idle_ttl`, returning their indexes. The caller closes their
+    /// at least `idle_ttl`, returning their tickets. The caller closes their
     /// sockets; the woken workers finish with a stale ticket and change
     /// nothing (invariant 1). Reserved slots are never reaped: recovery is
     /// closed after one request and install is one-time.
-    pub fn reap_idle(&mut self, now: Instant, idle_ttl: Duration) -> Vec<usize> {
+    pub fn reap_idle(&mut self, now: Instant, idle_ttl: Duration) -> Vec<SlotTicket> {
         let pool = self.pool;
         let mut reaped = Vec::new();
         for (index, slot) in self.slots.iter_mut().enumerate() {
@@ -227,19 +248,79 @@ impl Slots {
                 break;
             }
             if let SlotState::Running {
+                generation,
+                class,
                 phase: Phase::Idle { since },
-                ..
             } = slot
             {
                 if now.saturating_duration_since(*since) >= idle_ttl {
+                    reaped.push(SlotTicket {
+                        index,
+                        generation: *generation,
+                        class: *class,
+                    });
                     *slot = SlotState::Vacant;
-                    reaped.push(index);
                 }
             }
         }
         reaped
     }
 
+    /// Frees every pool slot whose in-flight request passed its `deadline`,
+    /// returning their tickets so the caller closes their sockets and records
+    /// one `Expired` retained failure each (#547 slice 5). The interrupted
+    /// operation's effect is unknown, so it is never released as clean. The
+    /// reserved slot is never expired automatically: install is one-time and
+    /// recovery is closed after one request.
+    pub fn reap_expired(&mut self, now: Instant) -> Vec<SlotTicket> {
+        let pool = self.pool;
+        let mut expired = Vec::new();
+        for (index, slot) in self.slots.iter_mut().enumerate() {
+            if index >= pool {
+                break;
+            }
+            if let SlotState::Running {
+                generation,
+                class,
+                phase: Phase::InFlight { deadline, .. },
+            } = slot
+            {
+                if now >= *deadline {
+                    expired.push(SlotTicket {
+                        index,
+                        generation: *generation,
+                        class: *class,
+                    });
+                    *slot = SlotState::Vacant;
+                }
+            }
+        }
+        expired
+    }
+
+    /// Frees one running slot whose generation matches, returning its ticket;
+    /// `None` when the slot is vacant or a newer worker owns it (#547 slice
+    /// 5). Unlike `finish`, this is operator-initiated: the caller closes the
+    /// socket and records `Reclaimed`. Any class may be reclaimed.
+    pub fn reclaim(&mut self, slot: usize, generation: u64) -> Option<SlotTicket> {
+        match self.slots.get(slot) {
+            Some(SlotState::Running {
+                generation: owned,
+                class,
+                ..
+            }) if *owned == generation => {
+                let ticket = SlotTicket {
+                    index: slot,
+                    generation: *owned,
+                    class: *class,
+                };
+                self.slots[slot] = SlotState::Vacant;
+                Some(ticket)
+            }
+            _ => None,
+        }
+    }
+
     /// Ends one worker: releases its slot and, for a retained failure, returns
     /// the already-described record for the caller to store. A ticket whose
     /// generation or class no longer matches changes nothing.
@@ -309,14 +390,17 @@ pub enum Disposition {
 }
 
 /// The recovery operation set the reserved slot serves once the daemon is
-/// ControlReady (#547 slice 3): close, workspace status, and the operator
-/// read model. Everything else is refused `Capacity` and the connection is
-/// closed. `backend.daemon.reclaim` joins this set in slice 5.
+/// ControlReady (#547 slice 3): close, workspace status, the operator read
+/// model and reclaim (#547 slice 5). Everything else is refused `Capacity`
+/// and the connection is closed.
 pub fn recovery_request(request: &layerfs_bridge::control::Request) -> bool {
     use layerfs_bridge::control::Request;
     match request {
         Request::Status(_) | Request::Unmount(_) => true,
-        Request::Sdk(call) => call.method == layerfs_bridge::control::OPERATION,
+        Request::Sdk(call) => {
+            call.method == layerfs_bridge::control::OPERATION
+                || call.method == layerfs_bridge::control::RECLAIM
+        }
         _ => false,
     }
 }
@@ -354,6 +438,10 @@ pub(super) struct State {
     pub startup: Startup,
     pub slots: Slots,
     pub retained: RetainedFailures,
+    /// One detached or joined worker per slot. A reclaimed or expired slot
+    /// detaches its stuck worker (its handle is dropped here) so the accept
+    /// loop never joins a worker that a Store/fence wait still holds.
+    pub workers: Vec<Option<JoinHandle<()>>>,
 }
 /// One real daemon application; Store/Overlay are initialized once and reused.
 pub struct Application {
@@ -366,10 +454,11 @@ pub struct Application {
     pub(super) export: Option<Arc<crate::control::ExportServing>>,
     pub(super) state: Mutex<State>,
     pub(super) changed: Condvar,
-    pub(super) workers: Mutex<Vec<Option<JoinHandle<()>>>>,
-    /// Per-slot socket fences: the accept loop closes one to reclaim an idle
-    /// slot, waking its worker parked in `receive()` (#547 slice 4).
-    pub(super) fences: Mutex<Vec<Option<transport::CloseHandle>>>,
+    /// Per-slot socket fences, keyed by the admission generation that owns
+    /// them: the accept loop closes one to reclaim an idle slot or to expire
+    /// an in-flight request (#547 slices 4-5). A generation key lets a stale
+    /// worker's late `clear_fence` skip a newer worker's fence.
+    pub(super) fences: Mutex<Vec<Option<(u64, transport::CloseHandle)>>>,
 }
 impl Application {
     pub(super) fn start(setup: DaemonSetup) -> Result<Arc<Self>, ApplicationError> {
@@ -524,9 +613,9 @@ impl Application {
                 },
                 slots: Slots::new(count),
                 retained: RetainedFailures::default(),
+                workers: (0..count).map(|_| None).collect(),
             }),
             changed: Condvar::new(),
-            workers: Mutex::new((0..count).map(|_| None).collect()),
             fences: Mutex::new((0..count).map(|_| None).collect()),
         }))
     }
diff --git a/engine/crates/layerfs-daemon/src/application/retained.rs b/engine/crates/layerfs-daemon/src/application/retained.rs
index c08a350cc7..ff7011dd4d 100644
--- a/engine/crates/layerfs-daemon/src/application/retained.rs
+++ b/engine/crates/layerfs-daemon/src/application/retained.rs
@@ -54,6 +54,41 @@ pub struct RetainedFailure {
 }
 
 impl RetainedFailure {
+    /// The record for a slot whose in-flight request passed
+    /// `request_deadline` (#547 slice 5): its effect is unknown, so it is
+    /// recorded rather than released, with `reply_sent: false`.
+    pub fn expired(ticket: super::owner::SlotTicket, owner: &str) -> Self {
+        Self::phase(ticket, owner, "Expired")
+    }
+
+    /// The record for a slot an operator reclaimed by
+    /// `backend.daemon.reclaim` (#547 slice 5): its in-flight effect is
+    /// unknown and its worker's socket is closed.
+    pub fn reclaimed(ticket: super::owner::SlotTicket, owner: &str) -> Self {
+        Self::phase(ticket, owner, "Reclaimed")
+    }
+
+    /// The common shape of an `Expired` or `Reclaimed` record: no decoded
+    /// request, no reply sent, and a bounded textual deciding cause.
+    fn phase(ticket: super::owner::SlotTicket, owner: &str, cause: &str) -> Self {
+        Self {
+            slot: ticket.index,
+            class: ticket.class.as_str(),
+            generation: ticket.generation,
+            owner: owner.to_owned(),
+            received_bytes: None,
+            request: None,
+            reply_sent: false,
+            cause: cause.to_owned(),
+            fence_error: None,
+            diagnostic_error: None,
+            age: Duration::ZERO,
+            expires_in: TTL,
+            retained: 0,
+            limit: LIMIT,
+        }
+    }
+
     /// The bounded record for `failure`, built outside the state lock.
     pub fn describe(
         failure: &ConnectionFailure,
diff --git a/engine/crates/layerfs-daemon/tests/daemon_slots.rs b/engine/crates/layerfs-daemon/tests/daemon_slots.rs
index 0455994619..56940529e5 100644
--- a/engine/crates/layerfs-daemon/tests/daemon_slots.rs
+++ b/engine/crates/layerfs-daemon/tests/daemon_slots.rs
@@ -18,8 +18,8 @@ mod host;
 
 use layerfs_bridge::{
     control::{
-        Answer, Call, CapacityModel, DaemonPhase, HelloRequest, Reply, Request, SdkCall, SdkValue,
-        OPERATION,
+        Answer, Call, CapacityModel, ControlCode, DaemonPhase, HelloRequest, Reply, Request,
+        SdkCall, SdkValue, OPERATION, RECLAIM,
     },
     daemon_setup::{DaemonLimits, DaemonSetup},
     duplex::{self, ChannelError, Connection},
@@ -70,7 +70,7 @@ fn private_directory(path: &Path) {
 }
 
 /// Starts the actual executable and returns it once it reports its listener.
-fn start(label: &str, handshake_ms: u32, idle_ttl_ms: u32) -> Daemon {
+fn start(label: &str, handshake_ms: u32, idle_ttl_ms: u32, request_deadline_ms: u32) -> Daemon {
     use std::sync::atomic::{AtomicU64, Ordering};
     static NEXT: AtomicU64 = AtomicU64::new(0);
     let directory = std::env::temp_dir().join(format!(
@@ -97,6 +97,7 @@ fn start(label: &str, handshake_ms: u32, idle_ttl_ms: u32) -> Daemon {
             connections: CONNECTIONS,
             handshake_ms,
             idle_ttl_ms,
+            request_deadline_ms,
             cache_bytes: 1024 * 1024,
             namespaces: 16,
             pager_kib: 1024,
@@ -231,7 +232,7 @@ fn wait_for_log(path: &Path, needle: &str, count: usize) {
 /// admitted, the pool is whole, and one log line names each failure (#540).
 #[test]
 fn undecodable_records_do_not_consume_control_slots() {
-    let daemon = start("slots", 2000, 60_000);
+    let daemon = start("slots", 2000, 60_000, 60_000);
     for round in 0..ROUNDS {
         let mut connection = connect(daemon.address);
         connection.send.send(DIE).unwrap();
@@ -282,7 +283,7 @@ fn undecodable_records_do_not_consume_control_slots() {
 /// `LAYERFS-CAPACITY` body instead of a bare EOF (#540).
 #[test]
 fn no_vacant_slot_writes_the_capacity_notice() {
-    let daemon = start("notice", 10_000, 60_000);
+    let daemon = start("notice", 10_000, 60_000, 60_000);
     let mut holders = Vec::new();
     for _ in 0..CONNECTIONS {
         holders.push(hello(daemon.address));
@@ -312,7 +313,7 @@ fn no_vacant_slot_writes_the_capacity_notice() {
     // No Store is installed, so the daemon is still starting.
     assert_eq!(
         notice.reason,
-        layerfs_bridge::duplex::admission::AdmissionReason::Starting
+        layerfs_bridge::duplex::admission::AdmissionReason::starting()
     );
     assert_eq!(notice.pool_vacant, 0);
     assert_eq!(notice.reserved_vacant, 0);
@@ -327,7 +328,7 @@ fn no_vacant_slot_writes_the_capacity_notice() {
 /// authenticated client is still admitted afterwards.
 #[test]
 fn concurrent_failures_do_not_deadlock_or_leak() {
-    let daemon = start("concurrent", 3000, 60_000);
+    let daemon = start("concurrent", 3000, 60_000, 60_000);
     let address = daemon.address;
     let mut threads = Vec::new();
     for _ in 0..16 {
@@ -392,7 +393,7 @@ fn concurrent_failures_do_not_deadlock_or_leak() {
 #[test]
 fn idle_clients_are_reclaimed_for_a_new_connection() {
     const IDLE_TTL_MS: u32 = 300;
-    let daemon = start("idle", 60_000, IDLE_TTL_MS);
+    let daemon = start("idle", 60_000, IDLE_TTL_MS, 60_000);
     let holders: Vec<Connection> = (0..CONNECTIONS).map(|_| hello(daemon.address)).collect();
     // Let every idle worker pass the TTL so the next admission reclaims one.
     thread::sleep(Duration::from_millis(u64::from(IDLE_TTL_MS) + 400));
@@ -471,6 +472,7 @@ fn start_ready(label: &str) -> Daemon {
             connections: CONNECTIONS,
             handshake_ms: 60_000,
             idle_ttl_ms: 60_000,
+            request_deadline_ms: 60_000,
             cache_bytes: 1024 * 1024,
             namespaces: 16,
             pager_kib: 1024,
@@ -638,3 +640,124 @@ fn reserved_slot_serves_recovery_and_refuses_everything_else() {
 
     drop(holders);
 }
+
+/// A worker whose decoded request passes `request_deadline` is expired: its
+/// slot is freed, its socket is closed (so the caller observes the loss), and
+/// one `Expired` retained failure is recorded (#547 slice 5). The interrupted
+/// operation's effect is unknown, so it is never released as clean.
+#[test]
+fn in_flight_expiry_records_expired_and_closes_the_socket() {
+    const DEADLINE_MS: u32 = 300;
+    let daemon = start("expiry", 60_000, 60_000, DEADLINE_MS);
+    // Park one worker in-flight: a Hello that waits for a Store that never
+    // installs blocks its worker in `hello()` past the request deadline.
+    let mut parked = connect(daemon.address);
+    parked
+        .send
+        .send(
+            &Call {
+                id: 1,
+                request: Request::Hello(HelloRequest {
+                    expected_instance: None,
+                    wait_for_store: true,
+                }),
+            }
+            .encode()
+            .unwrap(),
+        )
+        .unwrap();
+    thread::sleep(Duration::from_millis(u64::from(DEADLINE_MS) + 400));
+
+    // The capacity query's admission sweeps the expired slot first, so the
+    // read model already carries the Expired record.
+    let model = capacity(daemon.address);
+
+    // The parked worker's socket was closed by expiry: the caller observes
+    // EOF, never a silent hang.
+    assert!(parked.receive.receive().is_err());
+
+    let expired = model
+        .retained
+        .failures
+        .iter()
+        .find(|record| record.cause == "Expired")
+        .expect("no Expired retained failure");
+    assert_eq!(expired.class, "pool");
+    assert!(!expired.reply_sent);
+    assert!(expired.generation > 0);
+    // The slot was freed: only the query's own connection is running now.
+    assert_eq!(model.admission.pool_running, 1);
+    assert_eq!(model.admission.pool_vacant, i64::from(CONNECTIONS) - 1);
+}
+
+/// `backend.daemon.reclaim { slot, generation }` closes the named worker's
+/// socket, records `Reclaimed` and is one-shot: a second reclaim of the same
+/// generation answers `Invalid` (#547 slice 5).
+#[test]
+fn reclaim_closes_the_named_worker_and_answers_invalid_on_mismatch() {
+    let daemon = start("reclaim", 60_000, 60_000, 60_000);
+    // The first admitted worker: slot 0, generation 1.
+    let mut parked = connect(daemon.address);
+    parked
+        .send
+        .send(
+            &Call {
+                id: 1,
+                request: Request::Hello(HelloRequest {
+                    expected_instance: None,
+                    wait_for_store: true,
+                }),
+            }
+            .encode()
+            .unwrap(),
+        )
+        .unwrap();
+    thread::sleep(Duration::from_millis(200));
+
+    let reclaim = |slot: i64, generation: i64| {
+        let mut operator = connect(daemon.address);
+        operator
+            .send
+            .send(
+                &Call {
+                    id: 9,
+                    request: Request::Sdk(SdkCall {
+                        method: RECLAIM.to_owned(),
+                        args: SdkValue::map([
+                            ("slot", SdkValue::Int(slot)),
+                            ("generation", SdkValue::Int(generation)),
+                        ]),
+                    }),
+                }
+                .encode()
+                .unwrap(),
+            )
+            .unwrap();
+        Answer::decode(operator.receive.receive().unwrap())
+            .unwrap()
+            .reply
+    };
+
+    // The exact generation succeeds and frees the slot.
+    match reclaim(0, 1) {
+        Reply::Sdk(value) => {
+            assert_eq!(value.get_text("outcome"), Some("reclaimed"));
+            assert_eq!(value.get_int("slot"), Some(0));
+            assert_eq!(value.get_int("generation"), Some(1));
+        }
+        other => panic!("reclaim was not answered: {other:?}"),
+    }
+    // The parked worker's socket was closed by the reclaim.
+    assert!(parked.receive.receive().is_err());
+
+    // One-shot: the same generation now answers Invalid.
+    match reclaim(0, 1) {
+        Reply::Refused(refusal) => assert_eq!(refusal.code, ControlCode::Invalid),
+        other => panic!("stale reclaim was not refused: {other:?}"),
+    }
+    // A wrong generation on the now-reoccupied slot answers Invalid too.
+    match reclaim(0, 999) {
+        Reply::Refused(refusal) => assert_eq!(refusal.code, ControlCode::Invalid),
+        other => panic!("mismatched reclaim was not refused: {other:?}"),
+    }
+}
diff --git a/engine/crates/layerfs-daemon/tests/slot_state.rs b/engine/crates/layerfs-daemon/tests/slot_state.rs
index 89232e1405..d8c9f17bac 100644
--- a/engine/crates/layerfs-daemon/tests/slot_state.rs
+++ b/engine/crates/layerfs-daemon/tests/slot_state.rs
@@ -10,7 +10,7 @@ use layerfs_bridge::control::{Call, Reply, Request};
 use layerfs_bridge::duplex::{ChannelError, ChannelWork, PeerEndFailure, PeerEndPhase};
 use layerfs_daemon::application::internals::{
     disposition, recovery_request, AdmissionClass, Finished, RetainedFailure, RetainedFailures,
-    SlotClass, Slots, LIMIT, RESERVED_SLOTS, TTL,
+    SlotClass, SlotTicket, Slots, LIMIT, RESERVED_SLOTS, TTL,
 };
 use layerfs_daemon::application::{ApplicationError, ConnectionFailure};
 use layerfs_daemon::control::{Served, Success};
@@ -327,7 +327,7 @@ fn idle_and_busy_track_the_reapable_phase() {
         .is_empty());
     // Idle since `now`: reapable once `idle_ttl` has passed.
     assert!(slots.idle(ticket, now));
-    assert_eq!(slots.reap_idle(now + ttl, ttl), vec![ticket.index]);
+    assert_eq!(slots.reap_idle(now + ttl, ttl), vec![ticket]);
     // A fresh worker is Busy again and cannot be reaped while idle is false.
     let second = slots.admit(AdmissionClass::ControlPool).unwrap();
     assert_eq!(second.index, ticket.index);
@@ -358,7 +358,7 @@ fn reap_idle_frees_only_pool_slots_parked_past_ttl() {
     assert!(slots.idle(recent, now + ttl - Duration::from_millis(1)));
     assert!(slots.idle(reserved, now));
     let at = now + ttl;
-    assert_eq!(slots.reap_idle(at, ttl), vec![early.index]);
+    assert_eq!(slots.reap_idle(at, ttl), vec![early]);
     let counts = slots.counts();
     assert_eq!(counts.pool_vacant, 1);
     assert_eq!(counts.pool_running, 2);
@@ -375,7 +375,7 @@ fn reaped_slot_readmits_and_the_old_ticket_finishes_stale() {
     let ttl = Duration::from_secs(1);
     let first = slots.admit(AdmissionClass::ControlPool).unwrap();
     assert!(slots.idle(first, now));
-    assert_eq!(slots.reap_idle(now + ttl, ttl), vec![first.index]);
+    assert_eq!(slots.reap_idle(now + ttl, ttl), vec![first]);
     let second = slots.admit(AdmissionClass::ControlPool).unwrap();
     assert_eq!(second.index, first.index);
     assert!(second.generation > first.generation);
@@ -429,3 +429,111 @@ fn recovery_request_admits_only_the_recovery_set() {
         }
     )));
 }
+
+/// A slot marked in-flight past its deadline is expired; the reserved slot is
+/// never expired automatically (install is one-time, recovery closes after
+/// one request) (#547 slice 5).
+#[test]
+fn inflight_slots_expire_only_past_their_deadline() {
+    let mut slots = Slots::new(3 + RESERVED_SLOTS);
+    let now = Instant::now();
+    let early = slots.admit(AdmissionClass::ControlPool).unwrap();
+    let late = slots.admit(AdmissionClass::ControlPool).unwrap();
+    let reserved = slots.admit(AdmissionClass::Recovery).unwrap();
+    assert!(slots.inflight(early, now, now + Duration::from_secs(5)));
+    assert!(slots.inflight(late, now, now + Duration::from_secs(20)));
+    assert!(slots.inflight(reserved, now, now + Duration::from_secs(5)));
+    // Before any deadline, nothing expires.
+    assert!(slots.reap_expired(now + Duration::from_secs(1)).is_empty());
+    // At 5 s only `early` expired, with its exact ticket.
+    let expired = slots.reap_expired(now + Duration::from_secs(5));
+    assert_eq!(expired.len(), 1);
+    assert_eq!(expired[0], early);
+    // At 20 s `late` expires too; the reserved slot still stays.
+    let expired = slots.reap_expired(now + Duration::from_secs(20));
+    assert_eq!(expired.len(), 1);
+    assert_eq!(expired[0], late);
+    let counts = slots.counts();
+    assert_eq!(counts.pool_vacant, 3);
+    assert_eq!(counts.pool_running, 0);
+    assert_eq!(counts.reserved_running, 1);
+}
+
+/// A reaped in-flight slot is immediately reusable; the woken worker's stale
+/// finish cannot free the newer worker (invariant 1, #547 slice 5).
+#[test]
+fn expired_slot_readmits_and_the_old_ticket_finishes_stale() {
+    let mut slots = Slots::new(1 + RESERVED_SLOTS);
+    let now = Instant::now();
+    let first = slots.admit(AdmissionClass::ControlPool).unwrap();
+    assert!(slots.inflight(first, now, now + Duration::from_secs(1)));
+    assert_eq!(
+        slots.reap_expired(now + Duration::from_secs(1)),
+        vec![first]
+    );
+    let second = slots.admit(AdmissionClass::ControlPool).unwrap();
+    assert_eq!(second.index, first.index);
+    assert!(second.generation > first.generation);
+    assert!(matches!(slots.finish(first, Ok(()), None), Finished::Stale));
+    assert_eq!(slots.counts().pool_running, 1);
+    slots.finish(second, Ok(()), None);
+    assert_eq!(slots.counts().pool_vacant, 1);
+}
+
+/// Reclaim frees exactly the named generation, on any class, and is one-shot
+/// (#547 slice 5).
+#[test]
+fn reclaim_frees_the_matching_generation_only() {
+    let mut slots = Slots::new(2 + RESERVED_SLOTS);
+    let now = Instant::now();
+    let first = slots.admit(AdmissionClass::ControlPool).unwrap();
+    assert!(slots.inflight(first, now, now + Duration::from_secs(60)));
+    // A wrong generation changes nothing.
+    assert!(slots.reclaim(first.index, first.generation + 1).is_none());
+    assert_eq!(slots.counts().pool_running, 1);
+    // The right generation frees the slot and returns its exact ticket.
+    let ticket = slots.reclaim(first.index, first.generation).unwrap();
+    assert_eq!(ticket, first);
+    assert_eq!(slots.counts().pool_vacant, 2);
+    // One-shot: the same generation no longer matches.
+    assert!(slots.reclaim(first.index, first.generation).is_none());
+    // The reserved slot is reclaimable too.
+    let reserved = slots.admit(AdmissionClass::Recovery).unwrap();
+    assert!(slots.reclaim(reserved.index, reserved.generation).is_some());
+    assert_eq!(slots.counts().reserved_vacant, 1);
+}
+
+/// `Expired` and `Reclaimed` records carry their deciding cause, the slot and
+/// generation, and never claim a reply was sent (#547 slice 5).
+#[test]
+fn expired_and_reclaimed_records_name_their_disposition() {
+    let ticket = SlotTicket {
+        index: 2,
+        generation: 7,
+        class: SlotClass::Pool,
+    };
+    let expired = RetainedFailure::expired(ticket, OWNER);
+    assert_eq!(expired.slot, 2);
+    assert_eq!(expired.class, "pool");
+    assert_eq!(expired.generation, 7);
+    assert!(!expired.reply_sent);
+    assert_eq!(expired.cause, "Expired");
+    assert!(expired.received_bytes.is_none());
+    assert!(expired.request.is_none());
+    let reclaimed = RetainedFailure::reclaimed(ticket, OWNER);
+    assert_eq!(reclaimed.slot, 2);
+    assert!(!reclaimed.reply_sent);
+    assert_eq!(reclaimed.cause, "Reclaimed");
+}
+
+/// `recovery_request` admits reclaim alongside the operator read model
+/// (#547 slice 5).
+#[test]
+fn recovery_request_admits_reclaim() {
+    assert!(recovery_request(&Request::Sdk(
+        layerfs_bridge::control::SdkCall {
+            method: layerfs_bridge::control::RECLAIM.to_owned(),
+            args: layerfs_bridge::control::SdkValue::Null,
+        }
+    )));
+}