diff --git a/context/design.md b/context/design.md
index 7d1051f..35202c3 100644
--- a/context/design.md
+++ b/context/design.md
@@ -175,7 +175,7 @@ The algorithm's existing Ruby performance motivates the port. Rust performance c
 
 1. Record boundaries and reusable Rust conventions (implemented).
 2. Replace stackful execution with async-task and Crossbeam worker queues (implemented). Keep the coroutine prototype in its saved branch.
-3. Implement primitive task dispatch, scheduler ownership, cancellation and shutdown (implemented). Runtime-independent cooperative signals and deferred cancellation are implemented; cancellation-aware file/socket operations and transferable task ownership are implemented. Clock cancellation is implemented; exclusive file close and bounded submission-only cleanup are implemented; wider cleanup hooks remain future work. Higher-level task groups are outside the executor for now.
+3. Implement primitive task dispatch, scheduler ownership, cancellation and shutdown (implemented). Runtime-independent cooperative signals and deferred cancellation are implemented; cancellation-aware file/socket operations and transferable task ownership are implemented. Clock cancellation is implemented; exclusive file/socket close, directional socket shutdown, and bounded submission-only cleanup are implemented; file advice remains future work. Higher-level task groups are outside the executor for now.
 4. Establish minimal clock and I/O contracts with concrete consumers (implemented with Socket, File, Clock and a portable TCP example).
 5. Implement native readiness and the Tokio adapter, running the same consumers with both (implemented). Native sleep uses async-io until scheduler integration with `socketry-timers`.
 6. Implement io\_uring's owned-buffer lifecycle, socket/file operations, cancellation and runtime probing (implemented). Improve operation reuse, buffer registration and submission backpressure in subsequent work.
diff --git a/context/gaps.md b/context/gaps.md
index 0511ff4..a6dab97 100644
--- a/context/gaps.md
+++ b/context/gaps.md
@@ -8,7 +8,7 @@ Task cancellation currently drops the future before its next poll. An active pol
 
 `Cancellation` and `defer_cancel` now provide runtime-independent cooperative requests and deferred cleanup. Signals can be shared or linked through child signals to shut down the application or an individual service. They signal intent; task handles still confirm completion.
 
-File and socket I/O operations now accept these signals. Explicit file close omits cancellation, rejects outstanding use, and uses bounded cleanup admission; graceful scheduler shutdown drains accepted closes. Applications explicitly await tasks they want to finish; abandoning an owning task handle requests destruction. Graceful shutdown must keep workers, I/O services, and the scheduler available until draining completes. Cooperative cancellation has no automatic forced abort or internal escalation; a process supervisor handles SIGTERM followed by SIGKILL. Existing task cancellation and scheduler shutdown still destroy futures, so callers must request cooperative cancellation and await work first.
+File and socket I/O operations now accept these signals. Explicit file/socket close omits cancellation, rejects outstanding use, and uses bounded cleanup admission; socket shutdown retains the borrowed descriptor. Graceful scheduler shutdown drains accepted cleanup requests. Applications explicitly await tasks they want to finish; abandoning an owning task handle requests destruction. Graceful shutdown must keep workers, I/O services, and the scheduler available until draining completes. Cooperative cancellation has no automatic forced abort or internal escalation; a process supervisor handles SIGTERM followed by SIGKILL. Existing task cancellation and scheduler shutdown still destroy futures, so callers must request cooperative cancellation and await work first.
 
 ## 2. Higher-level task grouping is deferred
 
@@ -18,7 +18,7 @@ The executor does not expose a Barrier or require a group/owner argument. Librar
 
 ## 3. The I/O boundary is still narrow
 
-`Socket` uses one associated resource type for unconnected, bound, listening, connected, and datagram sockets. It exposes `socket_open`, `socket_bind`, syscall-like `socket_listen`, `socket_connect`, `socket_accept`, address inspection, reads, writes, and readiness. The initial API supports IPv4/IPv6 and stream/datagram sockets with optional TCP/UDP protocol selection. OS-resource import is a separate `SocketRegistration` capability. Socket close/shutdown, options, message/address APIs, and ancillary data remain future work. `File` opens associated resources and borrows them for seeking, truncation, synchronization, shared-cursor and positioned I/O. Explicit close/submission now consumes an exclusively owned handle, and optional `FileAllocate` supplies Linux allocation. Metadata hooks remain to implement. File/socket operations support cancellation and owned buffer views. `Clock` supplies monotonic time and cancellable absolute and relative waits.
+`Socket` uses one associated resource type for unconnected, bound, listening, connected, and datagram sockets. It exposes `socket_open`, `socket_bind`, syscall-like `socket_listen`, `socket_connect`, `socket_accept`, address inspection, reads, writes, and readiness. The initial API supports IPv4/IPv6 and stream/datagram sockets with optional TCP/UDP protocol selection. OS-resource import is a separate `SocketRegistration` capability. Socket close/shutdown and their admission-only variants are implemented; options, message/address APIs, and ancillary data remain future work. `File` opens associated resources and borrows them for seeking, truncation, synchronization, shared-cursor and positioned I/O. Explicit close/submission now consumes an exclusively owned handle, and optional `FileAllocate` supplies Linux allocation. Metadata hooks remain to implement. File/socket operations support cancellation and owned buffer views. `Clock` supplies monotonic time and cancellable absolute and relative waits.
 
 `Blocking` now adds bounded asynchronous admission for owned synchronous closures, implemented through dedicated native and Tokio pools. Existing file fallbacks still use backend pools. Socket options, message/address APIs, DNS, randomness, and processes remain outside the portable boundary. Deadline policies can now compose `Clock` and cancellation; higher-level timeout helpers remain separate. Ordinary synchronization uses futures and wakers above scheduler primitives. Expand capabilities as real consumers need them, keeping the application's implementation choice explicit. A broader boundary would also allow more operating-system behavior to be substituted in deterministic tests.
 
diff --git a/context/implementation.md b/context/implementation.md
index 5461af6..f4ba8b8 100644
--- a/context/implementation.md
+++ b/context/implementation.md
@@ -32,7 +32,7 @@ This guide describes the current implementation and its boundaries. Read [the de
 - The pool uses a mutex-protected queue, a condition variable for workers, and event listeners for asynchronous admission. Register listeners before inspecting capacity; never invoke wakers or destroy user captures/results under the queue lock.
 - Check cancellation before admission and again with abandonment before calling the closure. Queued cancellation need not complete until dequeued. Once started, preserve the real output; closures can cooperate explicitly using a captured signal. Catch job panics and release capacity even when abandoned result destructors panic.
 - Close admission and discard queued jobs at scheduler shutdown; started jobs retain their data and finish independently. Do not join pool threads or implicitly drain jobs in `Scheduler::run`. Applications await them before shutting down services. Partial thread startup failure closes the pool and releases already-started workers.
-- Existing file fallbacks stay on their backend pools. File close uses a separate cleanup pool built from the same internal implementation: accepted jobs survive abandonment/closure and graceful shutdown drains them. Each pool uses the configured `BlockingOptions` limits. Public pool sharing and idle thread retirement remain separate work.
+- Existing file fallbacks stay on their backend pools. File close and socket close/shutdown use a separate cleanup pool built from the same internal implementation: accepted jobs survive abandonment/closure and graceful shutdown drains them. Each pool uses the configured `BlockingOptions` limits. Public pool sharing and idle thread retirement remain separate work.
 
 ## Cooperative cancellation
 
@@ -58,7 +58,7 @@ This guide describes the current implementation and its boundaries. Read [the de
 ## I/O and runtime boundaries
 
 - `scheduler.rs` re-exports Socket, File, Interest and Clock from `scheduler/socket.rs`, `file.rs`, `interest.rs` and `clock.rs`. Operations return concrete Send futures; portable consumers receive the required capabilities.
-- `File` declares an associated resource and opens it with `FileOptions`; native/Tokio choose `Arc<std::fs::File>` internally. `Socket` creates TCP listeners and exposes address inspection; standard-library resource import belongs to `SocketRegistration`. Blocking truncation, synchronization, shared-cursor, and positioned file helpers live in `scheduler/file_io.rs`. `file_seek` accepts `std::io::SeekFrom` and returns the resulting offset through a cancellable blocking fallback on all current native backends and the Tokio runtime pool; await it and other cursor operations in order. `file_read`/`file_write` advance the OS cursor with no extra per-file state or locking; io\_uring represents current-position requests with `None` internally and encodes them as offset `-1`. Await cursor operations sequentially when ordering matters, and retain buffers through terminal completion as for positioned I/O.
+- `File` declares an associated resource and opens it with `FileOptions`; native/Tokio choose `Arc<std::fs::File>` internally. `Socket` creates IP socket descriptors with explicit bind/listen/connect and address inspection; standard-library resource import belongs to `SocketRegistration`. Blocking truncation, synchronization, shared-cursor, and positioned file helpers live in `scheduler/file_io.rs`. `file_seek` accepts `std::io::SeekFrom` and returns the resulting offset through a cancellable blocking fallback on all current native backends and the Tokio runtime pool; await it and other cursor operations in order. `file_read`/`file_write` advance the OS cursor with no extra per-file state or locking; io\_uring represents current-position requests with `None` internally and encodes them as offset `-1`. Await cursor operations sequentially when ordering matters, and retain buffers through terminal completion as for positioned I/O.
 - `FileAllocate: File` optionally reserves space with `AllocationMode::Extend`/`KeepSize`. Native readiness and Tokio implement it on Linux using `fallocate64` on their file pools; io\_uring probes optional `Fallocate` and falls back when absent. Other targets currently omit the implementation. Positive length and a range end within `i64` are required; data and cursor are preserved, and filesystem support errors remain runtime results. No resize/zero-write emulation, unsharing, or durability promise. The simulator reserves vector capacity within its 1 MiB bound.
 - `File::file_truncate` changes the logical length with zero-filled extension, requiring write access and preserving the cursor. Lengths must fit in `i64`. Readiness and Tokio use `set_len` on their file pools; io\_uring probes optional `Ftruncate` support and falls back to the blocking implementation when absent. Buffer-free ring requests share resource retention and cancellation with sync. Await related operations sequentially; truncation provides neither space reservation nor durability. The simulator uses its 1 MiB bound.
 - `File::file_sync` takes `SyncMode::All` or `SyncMode::Data` and waits for OS completion. io\_uring submits buffer-free `Fsync` requests through the same cancellation/completion machinery as transfers and requires opcode support at initialization. Readiness and Tokio use their existing file pools with `sync_all`/`sync_data`; the simulator completes immediately without modeling durability. Await preceding writes first; sync leaves the cursor unchanged and does not flush application buffers or synchronize directory entries. File ownership is retained through terminal completion even after waiter abandonment.
@@ -75,7 +75,7 @@ This guide describes the current implementation and its boundaries. Read [the de
 
 ## Next boundaries
 
-- `socketry-timers`, also available through `socketry::timers`, implements the independent io-event timer queue with batched insertion, constant-time cancellation and lazy compaction. Its optional `sharded` feature adds worker queues, cross-worker assistance and lazy sleep futures; the experimental inbox driver remains optional. The crate has no executor dependency. Scheduler integration, operation pools, registered buffers, general descriptor/UDP interfaces and overlapped Windows file operations remain planned. `Clock::now` and cancellable absolute/relative waits use async-io on native and the selected runtime on Tokio. Relative sleeps start on first poll, and overflow maps to `InvalidInput`. Paused Tokio clocks are preserved; native timer-engine replacement remains separate work.
+- `socketry-timers`, also available through `socketry::timers`, implements the independent io-event timer queue with batched insertion, constant-time cancellation and lazy compaction. Its optional `sharded` feature adds worker queues, cross-worker assistance and lazy sleep futures; the experimental inbox driver remains optional. The crate has no executor dependency. Scheduler integration, operation pools, registered buffers, socket options/message interfaces and overlapped Windows file operations remain planned. `Clock::now` and cancellable absolute/relative waits use async-io on native and the selected runtime on Tokio. Relative sleeps start on first poll, and overflow maps to `InvalidInput`. Paused Tokio clocks are preserved; native timer-engine replacement remains separate work.
 - Keep runtime requirements explicit; ordinary Future support does not provide another runtime's I/O or timers.
 - There is no non-Send task executor or scoped borrowing.
 - Do not introduce unsafe Send, stack migration or nested blocking waits.
@@ -100,3 +100,12 @@ Coverage measures source regions for native and Tokio schedulers on each support
 - All operation result notifications must follow release of backend file references, including cancellation and requests discarded before submission. io_uring still retains published requests until original terminal CQEs; cancellation CQEs cannot release them.
 - `examples/file_close.rs` explains cancellation clones, keeping the operation future alive through cancellation, completion races, and explicit close before propagating an I/O result. Dropping an operation future abandons completion and may leave backend work running; io_uring requests cancellation, but immediate close can still panic until backend access ends.
 - Close consumes the OS handle once and reports syscall errors without retrying. `_submit` execution errors are currently discarded. Ordinary resource destruction remains best effort; native close SQEs/CQE suppression and richer debug tracking are deferred.
+
+## Explicit socket lifecycle
+
+- `Socket` requires directional shutdown with `std::net::Shutdown`, consuming close, and `_submit` variants. Ordinary futures await OS completion; `_submit` awaits admission and discards execution errors. These lazy cleanup operations have no cancellation input.
+- Native and Tokio descriptors privately retain their persistent registration through `Arc`, without exposing `Clone` or a common file/socket resource trait. Readiness reads/writes borrow it; accepted shutdown jobs and io_uring operations retain it. Explicit close checks exclusivity before admission and panics on outstanding use; the cleanup job then deregisters and closes on its worker.
+- File and socket cleanup share `scheduler/cleanup.rs` and each scheduler's existing bounded cleanup pool. Admission, abandonment, error reporting, and graceful draining use the same contract. Native socket close/shutdown currently use this pool even with io_uring selected; native SQEs are deferred.
+- Await prior writes before `Shutdown::Write`, and await shutdown completion before explicit close. `_submit` supplies no completion fence. Completion notifications follow release of retained backend ownership. Request and await I/O cancellation before close; never equate dropping an operation future with completion.
+- Windows cleanup switches an exclusively owned, deregistered socket to blocking mode before closing, preserving positive linger behavior on the cleanup worker. Unix close is attempted once. `Drop` remains best effort.
+- The simulator models directional EOF and disabled writes immediately, without deferred cleanup. Its `_submit` methods still discard execution errors after accepting the request. `examples/socket_shutdown.rs` explains request/response half-close and final ownership transfer.
diff --git a/crates/executor/design.md b/crates/executor/design.md
index 87f558a..36e58ce 100644
--- a/crates/executor/design.md
+++ b/crates/executor/design.md
@@ -447,7 +447,7 @@ pub trait Clock: Send + Sync {
 }
 ```
 
-The associated file resource, truncation, synchronization, and both stateful and positioned file operations above are implemented. The initial socket contract has explicit creation, bind, listen, connect, accept, stream/datagram read and write, readiness, and local/peer address operations. `socket_listen` receives an existing socket and a backlog. Socket close, options, message/address I/O, and portable flags remain separate design work. Explicit file close consumes its handle and follows the shared admission contract below.
+The associated file resource, truncation, synchronization, and both stateful and positioned file operations above are implemented. The initial socket contract has explicit creation, bind, listen, connect, accept, stream/datagram read and write, readiness, and local/peer address operations. `socket_listen` receives an existing socket and a backlog. Socket close and directional shutdown use the cleanup contract below. Options, message/address I/O, and portable flags remain separate design work. Explicit file close consumes its handle and follows the shared admission contract below.
 
 Protocol buffering belongs above the descriptor-level scheduler interface. A protocol can use a specialized transport adapter or a generic buffering layer around `S::Socket`, with ordinary static dispatch as the default. Dynamic dispatch is an option when applications need runtime selection or type erasure; it should not add indirection to every primitive operation. The scheduler itself does not promise line, frame, or message buffering.
 
@@ -464,7 +464,7 @@ These rows identify extension capabilities; they do not require every scheduler
 
 ### Completion and explicit submission
 
-Cleanup and selected request operations have two forms. The file-close pair is implemented; socket close/shutdown and advice remain proposed:
+Cleanup and selected request operations have two forms. File close and socket close/shutdown pairs are implemented; advice remains proposed:
 
 | Await admission and completion | Await admission only | Resource input |
 | --- | --- | --- |
@@ -485,17 +485,25 @@ scheduler.file_close(file).await?; // Completed; execution errors reported.
 scheduler.file_close_submit(file).await?; // Accepted for execution.
 ```
 
-The operation pairs share an admission queue and job ownership mechanism; file close establishes the first implementation. Close consumes its handle, while shutdown and advice retain the borrowed resource's underlying backend ownership for the job. This contract does not require a separate per-fd retirement mechanism. Queue capacity and fairness remain implementation choices; bounding a scheduler queue does not itself bound the number of caller-owned futures waiting for admission.
+File and socket cleanup pairs share an admission queue and job ownership mechanism. Close consumes its handle, while shutdown and advice retain the borrowed resource's underlying backend ownership for the job. This contract does not require a separate per-fd retirement mechanism. Queue capacity and fairness remain implementation choices; bounding a scheduler queue does not itself bound the number of caller-owned futures waiting for admission.
 
 These methods omit cancellation inputs. Before acceptance, the admission future owns its inputs: dropping it, or rejecting admission, releases those inputs according to their ordinary destruction rules. For close, this means dropping the consumed handle; no asynchronous close job has been accepted. For current file handles, ordinary destruction releases ownership and lets the last backend reference perform best-effort synchronous destruction; it cannot report close errors. Constructing a close future does not change this fallback policy.
 
 Once the scheduler accepts the job, it must carry out the request even if the caller drops either form's future or its cancellation scope is cancelled. Acceptance is the ownership boundary even if the caller abandons the future before observing its result. Dropping a waiter after acceptance does not return a consumed handle or undo submission.
 
-Submitting work does not establish completion order. Explicit file close requires exclusive ownership: applications must finish all operations using the handle first. It never cancels, interrupts, or waits for other operations. Native backends use `Arc::try_unwrap` to transfer private ownership and panic on outstanding use, without a per-fd registry. An optional debug implementation may add operation tracking and diagnostics. This exclusivity check and kernel-resource retention remain active in release builds. Dependencies for future socket shutdown and registered-descriptor reuse must also be explicit. Graceful runtime draining must include outstanding submitted jobs even when they have no application waiter.
+Submitting work does not establish completion order. Explicit file close requires exclusive ownership: applications must finish all operations using the handle first. It never cancels, interrupts, or waits for other operations. Native backends use `Arc::try_unwrap` to transfer private ownership and panic on outstanding use, without a per-fd registry. An optional debug implementation may add operation tracking and diagnostics. This exclusivity check and kernel-resource retention remain active in release builds. Dependencies for socket shutdown and registered-descriptor reuse must also be explicit. Graceful runtime draining must include outstanding submitted jobs even when they have no application waiter.
 
 Dropping an in-flight io_uring operation requests cancellation, but its file and buffer remain owned by the selector until the original terminal completion. The caller needing an orderly close must request cancellation and continue awaiting the operation. Returning an operation result (including cancellation) releases its backend file ownership before waking the caller. This differs from dropping the operation future, which cannot await cleanup and gives up the completion notification. Started blocking file calls may run to completion despite cancellation. See the commented [file-close example](examples/file_close.rs).
 
-File close initially uses a bounded blocking cleanup pool for native and Tokio schedulers, including the io_uring scheduler. Both variants use the same path. This pool reuses the internal blocking-pool implementation but has separate workers from arbitrary `spawn_blocking` closures, so shutdown draining does not depend on a user closure that might itself destroy the scheduler. Both pools start lazily and each uses the configured `BlockingOptions` limits. Accepted close jobs survive waiter and scheduler destruction. Native shutdown/destruction outside future and blocking-pool workers waits for cleanup; native destruction on either kind of worker only requests shutdown. Explicit synchronous shutdown panics on either kind of worker to avoid waiting for its own completion. Tokio destruction requests shutdown, and its awaited `shutdown` drains cleanup. Standalone io_uring selectors own a cleanup pool drained by their shutdown; standalone readiness selectors use a process-wide bounded fallback, like their other file operations. The simulator completes close immediately. `_submit` execution errors are currently discarded; use ordinary close for diagnostics. Native close SQEs and CQE suppression remain later optimizations.
+File close initially uses a bounded blocking cleanup pool for native and Tokio schedulers, including the io_uring scheduler. Both variants use the same path. This pool reuses the internal blocking-pool implementation but has separate workers from arbitrary `spawn_blocking` closures, so shutdown draining does not depend on a user closure that might itself destroy the scheduler. Both pools start lazily and each uses the configured `BlockingOptions` limits. Accepted close jobs survive waiter and scheduler destruction. Native shutdown/destruction outside future and blocking-pool workers waits for cleanup; native destruction on either kind of worker only requests shutdown. Explicit synchronous shutdown panics on either kind of worker to avoid waiting for its own completion. Tokio destruction requests shutdown, and its awaited `shutdown` drains cleanup. Standalone io_uring selectors own a cleanup pool drained by their shutdown; standalone readiness selectors use a process-wide bounded fallback, like their other file operations. The simulator completes close immediately. `_submit` execution errors are currently discarded; use ordinary close for diagnostics. Native close/shutdown SQEs and CQE suppression remain later optimizations.
+
+`Socket::socket_shutdown` takes `std::net::Shutdown::{Read, Write, Both}` and borrows the descriptor. Await preceding writes before shutting down `Write`; the other direction can remain usable for the response. Repeated shutdown and I/O on disabled directions retain OS-specific outcomes. `socket_close` consumes the descriptor and rejects retained backend use before admission; it never coordinates or interrupts other operations. Deregistration and close run together on the cleanup worker. Deregistration errors are execution results where the backend exposes them; Tokio's Unix `AsyncFd::into_inner()` discards them internally. The ordinary future reports available execution errors; `_submit` discards them.
+
+Native and Tokio socket descriptors privately share their persistent registration through `Arc`. Ordinary readiness operations borrow it without cloning; io_uring and accepted shutdown jobs retain references until completion. Cleanup releases those references before notifying completion, allowing an immediately following close. `_submit` alone cannot establish this ordering: use ordinary shutdown when subsequent explicit close depends on its completion. The in-memory backend completes immediately while preserving admission-only error reporting for `_submit`.
+
+On Windows, explicit close switches the exclusively owned, deregistered socket to blocking mode on the cleanup worker so positive linger can finish rather than returning `WSAEWOULDBLOCK` with the socket still open. It preserves the configured linger policy; neither close nor shutdown guarantees remote application processing. See [Winsock close semantics](https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-closesocket). Ordinary descriptor destruction remains best effort.
+
+The commented [half-close example](examples/socket_shutdown.rs) sends a request, shuts down writes, receives a response, and closes only after all operations complete. Shared conformance tests cover directional shutdown, EOF, cancellation followed by close, invalid-state errors, and explicit submission; backend tests cover admission backpressure, waiter abandonment, shutdown draining, retained ownership, and Tokio runtime identity.
 
 Shutdown completion confirms the local shutdown operation, not remote receipt or processing. Advice completion confirms that the hint was processed, not that all requested data was prefetched; `WILLNEED` initiates nonblocking read-ahead. See [shutdown](https://man7.org/linux/man-pages/man2/shutdown.2.html) and [file advice](https://man7.org/linux/man-pages/man2/posix_fadvise.2.html).
 
@@ -849,9 +857,9 @@ The public contracts now cover associated resources and creation, owned buffer r
 
 Remaining design work includes:
 
-1. **Socket lifecycle and wider I/O:** explicit socket shutdown/close, message/address I/O, ancillary data, socket options, and extension boundaries.
+1. **Wider socket I/O:** message/address I/O, ancillary data, socket options, and extension boundaries. Explicit directional shutdown and close are implemented.
 2. **Blocking pool integration:** the public `Blocking` capability now provides bounded admission and owned completion. Existing file fallbacks still use backend pools; sharing, idle thread retirement, and further scheduling policies remain separate decisions.
-3. **Backend retirement:** account for all accepted work, including `_submit` jobs without application waiters. File-close cleanup is now accounted for; extend this to subsequent submission-only operations. Application completion remains explicit through owned task handles; there is no implicit graceful join of all spawned work.
+3. **Backend retirement:** account for all accepted work, including `_submit` jobs without application waiters. File close and socket close/shutdown cleanup are accounted for; extend this to subsequent submission-only operations. Application completion remains explicit through owned task handles; there is no implicit graceful join of all spawned work.
 4. **Timer engine integration:** the monotonic `Clock` and cancellable deadline contract are implemented. Native waits still use async-io, and the adapter uses its selected Tokio runtime, including paused time. Integrating `socketry-timers` into scheduler driving and parking remains separate work.
 
 Process creation/waiting and pipes remain useful optional extensions. Prepared linking, registered buffers, and pools remain separate later work.
@@ -882,13 +890,13 @@ The simulator has bounded in-memory file sizes and unbounded byte queues. It doe
 | Executor | Future workers, work stealing, primitive `Spawn`, and owning handles implemented. Await joins; move transfers ownership; drop abandons unfinished work. Task groups remain above the primitives. |
 | Cancellation | `Cancellation`, `Cancelled`, `defer_cancel`, and cancellation-aware file/socket operations implemented. Application shutdown is cooperative. |
 | Signal notifications | `Notify<Value>`, `Signals`, and optional Unix registration guards implemented for native and Tokio; each guard owns its dispatch task. |
-| Runtime lifetime | Application waits are explicit. Runtime shutdown abandons futures; backend retention protects submitted I/O. Accepted file-close submissions survive abandonment and are drained by graceful shutdown. |
+| Runtime lifetime | Application waits are explicit. Runtime shutdown abandons futures; backend retention protects submitted I/O. Accepted file-close and socket-close/shutdown submissions survive abandonment and are drained by graceful shutdown. |
 | Capabilities | Public `File`, `Socket`, `SocketRegistration`, `Clock`, `Spawn`, and `Blocking`; associated resources and initial file/socket creation implemented. Broader hooks remain design work. |
 | Blocking execution | Dedicated native/Tokio pools, configurable worker and admission limits, queued cancellation, and owning completion handles implemented. Started closures finish independently; graceful completion requires awaiting them. |
 | Linked operations | Prepared chains and `Linked` remain proposed; native linking, whole-chain blocking jobs, and readiness sequencing are not implemented. |
 | File operations | `file_open`, `file_seek`, `file_sync`, `file_truncate`, shared-cursor `file_read`/`file_write`, and explicit-offset `file_read_at`/`file_write_at` implemented with owned buffers. Optional `FileAllocate` implemented on Linux with `Extend` and `KeepSize`. `file_close`/`file_close_submit` implemented with exclusive ownership and bounded cleanup; metadata hooks pending. |
-| Submission and completion | File close and its `_submit` variant share bounded admission and persistent cleanup jobs. Socket shutdown/close, advice pairs, and safe CQE suppression remain proposed. The public blocking pool has its own bounded admission. |
-| Socket operations | `socket_open`, `socket_bind`, syscall-like `socket_listen`, `socket_connect`, `socket_accept`, address inspection, read/write, and readiness implemented for IPv4/IPv6 stream and datagram sockets. Message/address I/O and explicit socket close remain pending. |
+| Submission and completion | File close and socket close/shutdown pairs share bounded admission and persistent cleanup jobs. Advice pairs and safe CQE suppression remain proposed. The public blocking pool has its own bounded admission. |
+| Socket operations | `socket_open`, `socket_bind`, syscall-like `socket_listen`, `socket_connect`, `socket_accept`, address inspection, read/write, and readiness implemented for IPv4/IPv6 stream and datagram sockets. Directional shutdown and explicit close, with completion and admission-only forms, are implemented. Message/address I/O remains pending. |
 | Backends | Native readiness, blocking file fallbacks, Linux io_uring, and an adapter to an existing Tokio runtime implemented. Tokio uses `AsyncFd` on Unix and async-io readiness on Windows; the example simulator implements the public I/O traits. |
 | Buffers | Unsafe `IoBuf`/`IoBufMut`, `Vec<u8>`, `Box<[u8]>`, and owned `BufferSlice` implemented. Registered leases and pools remain proposed. |
 | Buffer forwarding | `with_previous_buffer` is hypothetical, with no selected implementation target. |
diff --git a/crates/executor/examples/interface_probe/simulated.rs b/crates/executor/examples/interface_probe/simulated.rs
index b40dfae..e02e71e 100644
--- a/crates/executor/examples/interface_probe/simulated.rs
+++ b/crates/executor/examples/interface_probe/simulated.rs
@@ -330,6 +330,49 @@ impl File for Scheduler {
 
 impl Socket for Scheduler {
     type Socket = SocketDescriptor;
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        drop(socket);
+        Ok(())
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        self.socket_close(socket).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        let (input, output) = match &*socket.state.lock().unwrap() {
+            SocketState::Connected(connection) => (
+                Arc::clone(&connection.input),
+                Arc::clone(&connection.output),
+            ),
+            _ => return Err(io::Error::from(io::ErrorKind::NotConnected)),
+        };
+        if matches!(how, std::net::Shutdown::Read | std::net::Shutdown::Both) {
+            let mut state = input.state.lock().unwrap();
+            state.reader_closed = true;
+            state.bytes.clear();
+        }
+        if matches!(how, std::net::Shutdown::Write | std::net::Shutdown::Both) {
+            output.state.lock().unwrap().writer_closed = true;
+        }
+        input.ready.notify(usize::MAX);
+        output.ready.notify(usize::MAX);
+        Ok(())
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        // This backend completes immediately, but submission still reports admission rather than the operation's result.
+        let _ = self.socket_shutdown(socket, how).await;
+        Ok(())
+    }
 
     async fn socket_open(
         &self,
@@ -559,7 +602,8 @@ impl Socket for Scheduler {
                         let ready = input.ready.listen();
                         {
                             let state = input.state.lock().unwrap();
-                            if !state.bytes.is_empty() || state.writer_closed {
+                            if !state.bytes.is_empty() || state.writer_closed || state.reader_closed
+                            {
                                 return Ok(());
                             }
                         }
@@ -605,7 +649,11 @@ impl Socket for Scheduler {
                     .len()
                     .min(self.read_limit)
                     .min(state.bytes.len());
-                if count > 0 || buffer.as_ref().is_empty() || state.writer_closed {
+                if count > 0
+                    || buffer.as_ref().is_empty()
+                    || state.writer_closed
+                    || state.reader_closed
+                {
                     for byte in &mut buffer.as_mut()[..count] {
                         *byte = state.bytes.pop_front().unwrap();
                     }
@@ -639,7 +687,7 @@ impl Socket for Scheduler {
         let count = buffer.as_ref().len().min(self.write_limit);
         {
             let mut state = output.state.lock().unwrap();
-            if state.reader_closed {
+            if state.reader_closed || state.writer_closed {
                 return (
                     Err(io::Error::from(io::ErrorKind::BrokenPipe).into()),
                     buffer,
diff --git a/crates/executor/examples/interface_probe/tests/socket_lifecycle.rs b/crates/executor/examples/interface_probe/tests/socket_lifecycle.rs
new file mode 100644
index 0000000..f194c47
--- /dev/null
+++ b/crates/executor/examples/interface_probe/tests/socket_lifecycle.rs
@@ -0,0 +1,139 @@
+// Released under the MIT License.
+// Copyright, 2026, by Samuel Williams.
+
+use socketry_executor::{Cancellation, OperationError, Socket, SocketDomain, SocketType};
+use std::net::Shutdown;
+
+async fn open<S: Socket>(scheduler: &S) -> S::Socket {
+    scheduler
+        .socket_open(
+            SocketDomain::Ipv4,
+            SocketType::Stream,
+            None,
+            &Cancellation::never(),
+        )
+        .await
+        .unwrap()
+}
+
+async fn pair<S: Socket>(scheduler: &S) -> (S::Socket, S::Socket) {
+    let never = Cancellation::never();
+    let listener = open(scheduler).await;
+    scheduler
+        .socket_bind(&listener, "127.0.0.1:0".parse().unwrap(), &never)
+        .await
+        .unwrap();
+    scheduler
+        .socket_listen(&listener, 16, &never)
+        .await
+        .unwrap();
+    let client = open(scheduler).await;
+    scheduler
+        .socket_connect(
+            &client,
+            scheduler.socket_local_addr(&listener).unwrap(),
+            &never,
+        )
+        .await
+        .unwrap();
+    let (server, _) = scheduler.socket_accept(&listener, &never).await.unwrap();
+    scheduler.socket_close(listener).await.unwrap();
+    (client, server)
+}
+
+pub async fn check<S: Socket>(scheduler: &S) {
+    let never = Cancellation::never();
+    let (client, server) = pair(scheduler).await;
+    // An unpolled shutdown must have no effect; the following write still succeeds.
+    drop(scheduler.socket_shutdown(&client, Shutdown::Both));
+    drop(scheduler.socket_shutdown_submit(&client, Shutdown::Both));
+    let (result, _) = scheduler.socket_write(&client, vec![42], &never).await;
+    assert_eq!(result.unwrap(), 1);
+    scheduler
+        .socket_shutdown(&client, Shutdown::Write)
+        .await
+        .unwrap();
+    let (result, buffer) = scheduler.socket_read(&server, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 1);
+    assert_eq!(buffer, vec![42]);
+    let (result, _) = scheduler.socket_read(&server, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 0);
+    let (result, _) = scheduler.socket_write(&client, vec![1], &never).await;
+    assert!(matches!(result, Err(OperationError::Io(_))));
+    // EOF in one direction does not prevent the response in the other direction.
+    let (result, _) = scheduler.socket_write(&server, vec![24], &never).await;
+    assert_eq!(result.unwrap(), 1);
+    scheduler
+        .socket_shutdown(&server, Shutdown::Write)
+        .await
+        .unwrap();
+    let (result, buffer) = scheduler.socket_read(&client, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 1);
+    assert_eq!(buffer, vec![24]);
+    let (result, _) = scheduler.socket_read(&client, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 0);
+    scheduler.socket_close(client).await.unwrap();
+    scheduler.socket_close(server).await.unwrap();
+
+    for how in [Shutdown::Read, Shutdown::Both] {
+        let (client, server) = pair(scheduler).await;
+        let mut read = Box::pin(scheduler.socket_read(&server, vec![0], &never));
+        assert!(futures_lite::future::poll_once(&mut read).await.is_none());
+        scheduler.socket_shutdown(&server, how).await.unwrap();
+        // Directional shutdown must also wake a receive that was already waiting.
+        let (result, _) = read.await;
+        // Some platforms return EOF; Winsock can report WSAESHUTDOWN.
+        assert!(matches!(result, Ok(0) | Err(OperationError::Io(_))));
+        if how == Shutdown::Read {
+            let (result, _) = scheduler.socket_write(&server, vec![17], &never).await;
+            assert_eq!(result.unwrap(), 1);
+            let (result, buffer) = scheduler.socket_read(&client, vec![0], &never).await;
+            assert_eq!(result.unwrap(), 1);
+            assert_eq!(buffer, vec![17]);
+        }
+        scheduler.socket_close(client).await.unwrap();
+        scheduler.socket_close(server).await.unwrap();
+    }
+
+    let (client, server) = pair(scheduler).await;
+    scheduler
+        .socket_shutdown_submit(&server, Shutdown::Write)
+        .await
+        .unwrap();
+    // Admission alone cannot authorize explicit close of the same socket. Ordinary Drop lets the accepted job retain it until completion.
+    drop(server);
+    let (result, _) = scheduler.socket_read(&client, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 0);
+    scheduler.socket_close(client).await.unwrap();
+
+    let (client, server) = pair(scheduler).await;
+    scheduler.socket_close_submit(server).await.unwrap();
+    let (result, _) = scheduler.socket_read(&client, vec![0], &never).await;
+    assert_eq!(result.unwrap(), 0);
+    scheduler.socket_close(client).await.unwrap();
+
+    let (client, server) = pair(scheduler).await;
+    let stopping = Cancellation::new();
+    let mut read = Box::pin(scheduler.socket_read(&client, vec![0; 8], &stopping));
+    assert!(futures_lite::future::poll_once(&mut read).await.is_none());
+    stopping.cancel();
+    let (result, _) = read.await;
+    assert!(matches!(result, Err(OperationError::Cancelled(_))));
+    // Returning cancellation must release backend ownership before waking this close caller.
+    scheduler.socket_close(client).await.unwrap();
+    scheduler.socket_close(server).await.unwrap();
+
+    let socket = open(scheduler).await;
+    assert!(
+        scheduler
+            .socket_shutdown(&socket, Shutdown::Write)
+            .await
+            .is_err()
+    );
+    // Submission reports admission, so the same OS error is not returned here.
+    scheduler
+        .socket_shutdown_submit(&socket, Shutdown::Write)
+        .await
+        .unwrap();
+    drop(socket);
+}
diff --git a/crates/executor/examples/socket_shutdown.rs b/crates/executor/examples/socket_shutdown.rs
new file mode 100644
index 0000000..1fbc372
--- /dev/null
+++ b/crates/executor/examples/socket_shutdown.rs
@@ -0,0 +1,133 @@
+// Released under the MIT License.
+// Copyright, 2026, by Samuel Williams.
+
+//! A small loopback request/response exchange framed by TCP half-close. Run with `cargo run -p socketry-executor --example socket_shutdown --features tokio` to exercise both native and Tokio backends.
+
+#[cfg(any(feature = "native", feature = "tokio"))]
+use socketry_executor::{Cancellation, IoBuf, Socket, SocketDomain, SocketType};
+use std::io;
+#[cfg(any(feature = "native", feature = "tokio"))]
+use std::net::Shutdown;
+
+#[cfg(any(feature = "native", feature = "tokio"))]
+async fn send_all<S: Socket>(
+    scheduler: &S,
+    socket: &S::Socket,
+    mut buffer: Vec<u8>,
+    cancellation: &Cancellation,
+) -> io::Result<()> {
+    let mut offset = 0;
+    while offset < buffer.len() {
+        // A syscall can write only part of the requested bytes. The owned slice keeps the original allocation and returns it when the operation completes.
+        let (result, remaining) = scheduler
+            .socket_write(socket, buffer.slice(offset..), cancellation)
+            .await;
+        buffer = remaining.into_inner();
+        let written = result?;
+        if written == 0 {
+            return Err(io::ErrorKind::WriteZero.into());
+        }
+        offset += written;
+    }
+    Ok(())
+}
+
+#[cfg(any(feature = "native", feature = "tokio"))]
+async fn receive_all<S: Socket>(
+    scheduler: &S,
+    socket: &S::Socket,
+    cancellation: &Cancellation,
+) -> io::Result<Vec<u8>> {
+    let mut received = Vec::new();
+    let mut buffer = vec![0; 128];
+    loop {
+        let (result, returned) = scheduler.socket_read(socket, buffer, cancellation).await;
+        buffer = returned;
+        let count = result?;
+        if count == 0 {
+            // A nonempty TCP receive buffer returning zero means the peer has finished sending.
+            return Ok(received);
+        }
+        received.extend_from_slice(&buffer[..count]);
+    }
+}
+
+#[cfg(any(feature = "native", feature = "tokio"))]
+async fn exchange<S: Socket>(scheduler: &S) -> io::Result<()> {
+    // This example deliberately runs to completion. An application can pass its cooperative shutdown signal to reads and writes instead.
+    let cancellation = Cancellation::never();
+    let listener = scheduler
+        .socket_open(SocketDomain::Ipv4, SocketType::Stream, None, &cancellation)
+        .await?;
+    scheduler
+        .socket_bind(&listener, "127.0.0.1:0".parse().unwrap(), &cancellation)
+        .await?;
+    scheduler
+        .socket_listen(&listener, 16, &cancellation)
+        .await?;
+    let client = scheduler
+        .socket_open(SocketDomain::Ipv4, SocketType::Stream, None, &cancellation)
+        .await?;
+    scheduler
+        .socket_connect(
+            &client,
+            scheduler.socket_local_addr(&listener)?,
+            &cancellation,
+        )
+        .await?;
+    let (server, _) = scheduler.socket_accept(&listener, &cancellation).await?;
+    scheduler.socket_close(listener).await?;
+
+    // Both peers live in this one future for clarity. The small request fits in the socket buffer; real servers normally handle connections in separate tasks.
+    send_all(
+        scheduler,
+        &client,
+        b"Hello, half-close!".to_vec(),
+        &cancellation,
+    )
+    .await?;
+    // Finish the write operation first, then announce EOF in the client-to-server direction. This borrows `client`: it remains usable for receiving the response.
+    scheduler.socket_shutdown(&client, Shutdown::Write).await?;
+    let request = receive_all(scheduler, &server, &cancellation).await?;
+
+    // Receiving EOF did not close the server's sending direction. Echo the completed request and announce the end of the response.
+    send_all(scheduler, &server, request, &cancellation).await?;
+    scheduler.socket_shutdown(&server, Shutdown::Write).await?;
+    let response = receive_all(scheduler, &client, &cancellation).await?;
+    assert_eq!(response, b"Hello, half-close!");
+
+    // Every operation above has completed, so both descriptors can now be moved into close. Close consumes ownership; shutdown only disables the requested directions.
+    // If cancelling an operation, request cancellation and keep awaiting it before closing. Dropping its future can leave kernel work in flight, and explicit close may then panic.
+    scheduler.socket_close(client).await?;
+    scheduler.socket_close(server).await?;
+    // Use `socket_close_submit(socket).await` when only acceptance matters. Do not replace the shutdowns above with `_submit`: the next step depends on their completion.
+    Ok(())
+}
+
+fn main() -> io::Result<()> {
+    #[cfg(feature = "native")]
+    {
+        let scheduler = socketry_executor::Scheduler::with_workers(1)?;
+        // block_on drives this root future until the exchange completes. The same generic function works with either scheduler implementation.
+        scheduler.block_on(exchange(&scheduler))?;
+    }
+    #[cfg(feature = "tokio")]
+    {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()?;
+        let scheduler =
+            socketry_executor::scheduler::tokio::Scheduler::new(runtime.handle().clone());
+        runtime.block_on(exchange(&scheduler))?;
+        runtime.block_on(scheduler.shutdown());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn half_close_exchange_completes_on_enabled_backends() {
+        super::main().unwrap();
+    }
+}
diff --git a/crates/executor/readme.md b/crates/executor/readme.md
index 51f1e0d..7a3bf51 100644
--- a/crates/executor/readme.md
+++ b/crates/executor/readme.md
@@ -284,6 +284,22 @@ cargo run -p socketry-executor --example file_close --features tokio
 
 Both close forms initially use a dedicated bounded cleanup pool, separate from arbitrary `spawn_blocking` closures. Each pool uses the configured `BlockingOptions` limits and starts workers lazily. Native io_uring close SQEs are a later optimization.
 
+### Socket shutdown and close
+
+`socket_shutdown(&socket, std::net::Shutdown::Write).await` finishes the local sending direction while keeping the socket available to receive a response. Use `Read` or `Both` for the other shutdown modes. Await preceding writes first. Shutdown borrows the descriptor; `socket_close(socket).await` consumes it and reports OS completion. Neither operation accepts cancellation. Repeated shutdown and later I/O follow platform-specific behavior, and local completion does not confirm remote application receipt.
+
+Both have exceptional `_submit` forms which await only bounded admission. Accepted work survives abandonment and shares the dedicated cleanup pool and graceful shutdown draining used for file close. Execution errors are discarded by `_submit`. Await ordinary shutdown before explicitly closing the same socket: submission alone does not establish completion, and retained backend use causes close to panic. No per-descriptor operation registry or implicit cancellation/join is added.
+
+Socket descriptors remain non-`Clone`; private references retain the registration for submitted shutdown and io_uring work. Deregistration and close execute on the cleanup worker. On Windows, the exclusively owned socket is made blocking there before close to accommodate positive linger. Ordinary `Drop` remains best-effort cleanup without joining backend work.
+
+The [commented half-close example](examples/socket_shutdown.rs) demonstrates a request followed by write shutdown, a response in the opposite direction, and explicit close after completion:
+
+```sh
+cargo run -p socketry-executor --example socket_shutdown --features tokio
+```
+
+External `Socket` implementations must provide `socket_shutdown`, `socket_shutdown_submit`, `socket_close`, and `socket_close_submit`.
+
 ### Cancellation and shutdown
 
 Pre-cancelled file/socket operations do not start. Readiness waits stop on cancellation. io\_uring requests cancellation but waits for the original completion before returning the buffer; successful progress wins a racing cancellation. Blocking jobs check cancellation again on their pool worker, before execution; a running syscall returns its actual result.
diff --git a/crates/executor/src/scheduler.rs b/crates/executor/src/scheduler.rs
index c1d92f3..e6c12c6 100644
--- a/crates/executor/src/scheduler.rs
+++ b/crates/executor/src/scheduler.rs
@@ -6,6 +6,8 @@
 pub mod selector;
 pub mod socketry;
 
+#[cfg(any(feature = "native", feature = "tokio"))]
+mod cleanup;
 #[cfg(any(feature = "native", feature = "tokio"))]
 mod file_close;
 #[cfg(any(feature = "native", feature = "tokio"))]
diff --git a/crates/executor/src/scheduler/cleanup.rs b/crates/executor/src/scheduler/cleanup.rs
new file mode 100644
index 0000000..9102249
--- /dev/null
+++ b/crates/executor/src/scheduler/cleanup.rs
@@ -0,0 +1,52 @@
+// Released under the MIT License.
+// Copyright, 2026, by Samuel Williams.
+
+use crate::{BlockingError, TaskError, blocking_pool::BlockingPool};
+use futures_channel::oneshot;
+use std::io;
+
+pub(crate) type Completion = oneshot::Receiver<Result<io::Result<()>, TaskError>>;
+
+pub(crate) async fn run(
+    pool: &BlockingPool,
+    operation: impl FnOnce() -> io::Result<()> + Send + 'static,
+) -> io::Result<()> {
+    wait(submit(pool, operation).await?).await
+}
+
+pub(crate) async fn submit_only(
+    pool: &BlockingPool,
+    operation: impl FnOnce() -> io::Result<()> + Send + 'static,
+) -> io::Result<()> {
+    submit(pool, operation).await.map(drop)
+}
+
+pub(crate) async fn submit(
+    pool: &BlockingPool,
+    operation: impl FnOnce() -> io::Result<()> + Send + 'static,
+) -> io::Result<Completion> {
+    pool.submit(operation).await.map_err(|error| match error {
+        BlockingError::Io(error) => error,
+        BlockingError::SchedulerClosed => {
+            io::Error::new(io::ErrorKind::BrokenPipe, "scheduler is closed")
+        }
+        BlockingError::Cancelled => unreachable!("cleanup admission has no cancellation input"),
+    })
+}
+
+pub(crate) async fn wait(completion: Completion) -> io::Result<()> {
+    completion
+        .await
+        .map_err(io::Error::other)?
+        .map_err(|error| io::Error::other(error.to_string()))?
+}
+
+// Direct readiness selector consumers use a process-wide fallback, like their existing file operations. Scheduler consumers instead use their own bounded pool and shutdown accounting.
+#[cfg(feature = "native")]
+pub(crate) fn standalone_pool() -> &'static BlockingPool {
+    static POOL: std::sync::OnceLock<BlockingPool> = std::sync::OnceLock::new();
+    POOL.get_or_init(|| BlockingPool::new(crate::BlockingOptions::default()))
+}
+
+#[cfg(test)]
+pub(crate) mod tests;
diff --git a/crates/executor/src/scheduler/cleanup/tests.rs b/crates/executor/src/scheduler/cleanup/tests.rs
new file mode 100644
index 0000000..23ec19a
--- /dev/null
+++ b/crates/executor/src/scheduler/cleanup/tests.rs
@@ -0,0 +1,144 @@
+// Released under the MIT License.
+// Copyright, 2026, by Samuel Williams.
+
+use super::*;
+use crate::SocketRegistration;
+use futures_lite::future::poll_once;
+use std::{
+    io::Read,
+    net::{Shutdown, TcpListener, TcpStream},
+    sync::mpsc,
+    time::Duration,
+};
+
+const TIMEOUT: Duration = Duration::from_secs(10);
+
+fn pair<S: SocketRegistration>(scheduler: &S) -> (S::Socket, TcpStream) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
+    let (peer, _) = listener.accept().unwrap();
+    peer.set_read_timeout(Some(TIMEOUT)).unwrap();
+    (scheduler.register_socket(client).unwrap(), peer)
+}
+
+// A blocked worker makes admission and execution observably different. Each caller supplies a pool with one worker and capacity two.
+pub(crate) async fn accepted_socket_jobs<S: SocketRegistration>(
+    scheduler: &S,
+    pool: &BlockingPool,
+    mode: u8,
+) {
+    let (socket, mut peer) = pair(scheduler);
+    let (started, starting) = mpsc::channel();
+    let (release, released) = mpsc::channel();
+    drop(
+        pool.submit(move || {
+            started.send(()).unwrap();
+            released.recv_timeout(TIMEOUT).unwrap();
+        })
+        .await
+        .unwrap(),
+    );
+    starting.recv_timeout(TIMEOUT).unwrap();
+
+    match mode {
+        0 => {
+            let mut close = Box::pin(scheduler.socket_close(socket));
+            assert!(poll_once(&mut close).await.is_none());
+            drop(close);
+        }
+        1 => scheduler.socket_close_submit(socket).await.unwrap(),
+        2 => {
+            let mut shutdown = Box::pin(scheduler.socket_shutdown(&socket, Shutdown::Write));
+            assert!(poll_once(&mut shutdown).await.is_none());
+            drop(shutdown);
+            drop(socket);
+        }
+        3 => {
+            scheduler
+                .socket_shutdown_submit(&socket, Shutdown::Write)
+                .await
+                .unwrap();
+            drop(socket);
+        }
+        _ => unreachable!(),
+    }
+    peer.set_nonblocking(true).unwrap();
+    assert_eq!(
+        peer.peek(&mut [0]).unwrap_err().kind(),
+        io::ErrorKind::WouldBlock
+    );
+    peer.set_nonblocking(false).unwrap();
+    // Closing admission must retain every accepted job, including ones whose waiter or original descriptor has disappeared.
+    pool.close();
+    let mut draining = Box::pin(pool.wait_submitted());
+    assert!(poll_once(&mut draining).await.is_none());
+    release.send(()).unwrap();
+    draining.await;
+    assert_eq!(peer.read(&mut [0]).unwrap(), 0);
+}
+
+pub(crate) async fn saturated_socket_admission<S: SocketRegistration>(
+    scheduler: &S,
+    pool: &BlockingPool,
+) {
+    let (socket, mut peer) = pair(scheduler);
+    let (started, starting) = mpsc::channel();
+    let (release, released) = mpsc::channel();
+    drop(
+        pool.submit(move || {
+            started.send(()).unwrap();
+            released.recv_timeout(TIMEOUT).unwrap();
+        })
+        .await
+        .unwrap(),
+    );
+    starting.recv_timeout(TIMEOUT).unwrap();
+    let mut admission = Box::pin(scheduler.socket_shutdown_submit(&socket, Shutdown::Both));
+    assert!(poll_once(&mut admission).await.is_none());
+    drop(admission);
+    release.send(()).unwrap();
+    pool.wait_submitted().await;
+    // Abandonment before admission must not execute shutdown.
+    let (result, _) = scheduler
+        .socket_write(&socket, vec![5], &crate::Cancellation::never())
+        .await;
+    assert_eq!(result.unwrap(), 1);
+    let mut byte = [0];
+    peer.read_exact(&mut byte).unwrap();
+    assert_eq!(byte, [5]);
+    scheduler.socket_close(socket).await.unwrap();
+}
+
+pub(crate) async fn closed_socket_admission<S: SocketRegistration>(
+    scheduler: &S,
+    pool: &BlockingPool,
+) {
+    let (a, _peer_a) = pair(scheduler);
+    let (b, _peer_b) = pair(scheduler);
+    let (c, _peer_c) = pair(scheduler);
+    pool.close();
+    assert_eq!(
+        scheduler
+            .socket_shutdown(&a, Shutdown::Write)
+            .await
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::BrokenPipe
+    );
+    assert_eq!(
+        scheduler
+            .socket_shutdown_submit(&a, Shutdown::Write)
+            .await
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::BrokenPipe
+    );
+    assert_eq!(
+        scheduler.socket_close(b).await.unwrap_err().kind(),
+        io::ErrorKind::BrokenPipe
+    );
+    assert_eq!(
+        scheduler.socket_close_submit(c).await.unwrap_err().kind(),
+        io::ErrorKind::BrokenPipe
+    );
+}
diff --git a/crates/executor/src/scheduler/file_close.rs b/crates/executor/src/scheduler/file_close.rs
index f313457..09b4476 100644
--- a/crates/executor/src/scheduler/file_close.rs
+++ b/crates/executor/src/scheduler/file_close.rs
@@ -1,12 +1,10 @@
 // Released under the MIT License.
 // Copyright, 2026, by Samuel Williams.
 
-use crate::{BlockingError, FileDescriptor, TaskError, blocking_pool::BlockingPool};
-use futures_channel::oneshot;
+use super::cleanup::{self, Completion, wait};
+use crate::{FileDescriptor, blocking_pool::BlockingPool};
 use std::io;
 
-type Completion = oneshot::Receiver<Result<io::Result<()>, TaskError>>;
-
 pub(crate) async fn close(pool: &BlockingPool, file: FileDescriptor) -> io::Result<()> {
     wait(submit(pool, file).await?).await
 }
@@ -18,29 +16,7 @@ pub(crate) async fn submit_only(pool: &BlockingPool, file: FileDescriptor) -> io
 async fn submit(pool: &BlockingPool, file: FileDescriptor) -> io::Result<Completion> {
     // Establish exclusivity before admission. Never wait for or interrupt other uses of this handle.
     let file = file.into_file();
-    pool.submit(move || super::file_io::close(file))
-        .await
-        .map_err(|error| match error {
-            BlockingError::Io(error) => error,
-            BlockingError::SchedulerClosed => {
-                io::Error::new(io::ErrorKind::BrokenPipe, "scheduler is closed")
-            }
-            BlockingError::Cancelled => unreachable!("close admission has no cancellation input"),
-        })
-}
-
-async fn wait(completion: Completion) -> io::Result<()> {
-    completion
-        .await
-        .map_err(io::Error::other)?
-        .map_err(|error| io::Error::other(error.to_string()))?
-}
-
-// Direct readiness selector consumers use a process-wide fallback, like their existing file operations. Scheduler consumers instead use their own bounded pool and shutdown accounting.
-#[cfg(feature = "native")]
-pub(crate) fn standalone_pool() -> &'static BlockingPool {
-    static POOL: std::sync::OnceLock<BlockingPool> = std::sync::OnceLock::new();
-    POOL.get_or_init(|| BlockingPool::new(crate::BlockingOptions::default()))
+    cleanup::submit(pool, move || super::file_io::close(file)).await
 }
 
 #[cfg(test)]
diff --git a/crates/executor/src/scheduler/file_close/tests.rs b/crates/executor/src/scheduler/file_close/tests.rs
index e39ca7e..1eafd06 100644
--- a/crates/executor/src/scheduler/file_close/tests.rs
+++ b/crates/executor/src/scheduler/file_close/tests.rs
@@ -2,7 +2,8 @@
 // Copyright, 2026, by Samuel Williams.
 
 use super::*;
-use crate::BlockingOptions;
+use crate::{BlockingOptions, TaskError};
+use futures_channel::oneshot;
 use futures_lite::future::{block_on, poll_once};
 use std::{panic::AssertUnwindSafe, sync::Arc};
 
diff --git a/crates/executor/src/scheduler/selector/io_uring.rs b/crates/executor/src/scheduler/selector/io_uring.rs
index e68876f..f870584 100644
--- a/crates/executor/src/scheduler/selector/io_uring.rs
+++ b/crates/executor/src/scheduler/selector/io_uring.rs
@@ -673,6 +673,42 @@ impl SocketRegistration for Selector {
 impl SocketOperations for Selector {
     type Socket = SocketDescriptor;
 
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        self.check_open()?;
+        crate::scheduler::cleanup::run(&self.controller.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        self.check_open()?;
+        crate::scheduler::cleanup::submit_only(&self.controller.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.check_open()?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::run(&self.controller.cleanup, move || {
+            socket.get_ref().shutdown(how)
+        })
+        .await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.check_open()?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::submit_only(&self.controller.cleanup, move || {
+            socket.get_ref().shutdown(how)
+        })
+        .await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
diff --git a/crates/executor/src/scheduler/selector/io_uring/tests.rs b/crates/executor/src/scheduler/selector/io_uring/tests.rs
index 5a4be1e..6a23eea 100644
--- a/crates/executor/src/scheduler/selector/io_uring/tests.rs
+++ b/crates/executor/src/scheduler/selector/io_uring/tests.rs
@@ -69,7 +69,44 @@ fn completion_releases_file_ownership_before_waking_close_callers() {
 fn close_rejects_a_stopped_selector() {
     async_io::block_on(async {
         let selector = Selector::new().unwrap();
+        let socket = || {
+            SocketDescriptor::new(
+                socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(),
+            )
+            .unwrap()
+        };
+        let first = socket();
+        let second = socket();
+        let third = socket();
         selector.shutdown().await;
+        assert_eq!(
+            selector
+                .socket_shutdown(&first, std::net::Shutdown::Write)
+                .await
+                .unwrap_err()
+                .kind(),
+            io::ErrorKind::BrokenPipe
+        );
+        assert_eq!(
+            selector
+                .socket_shutdown_submit(&first, std::net::Shutdown::Write)
+                .await
+                .unwrap_err()
+                .kind(),
+            io::ErrorKind::BrokenPipe
+        );
+        assert_eq!(
+            selector.socket_close(second).await.unwrap_err().kind(),
+            io::ErrorKind::BrokenPipe
+        );
+        assert_eq!(
+            selector
+                .socket_close_submit(third)
+                .await
+                .unwrap_err()
+                .kind(),
+            io::ErrorKind::BrokenPipe
+        );
         assert_eq!(
             selector
                 .file_close(tempfile::tempfile().unwrap().into())
diff --git a/crates/executor/src/scheduler/selector/readiness.rs b/crates/executor/src/scheduler/selector/readiness.rs
index 95d2d3f..ec58156 100644
--- a/crates/executor/src/scheduler/selector/readiness.rs
+++ b/crates/executor/src/scheduler/selector/readiness.rs
@@ -67,6 +67,48 @@ impl SocketRegistration for Selector {
 
 impl SocketOperations for Selector {
     type Socket = SocketDescriptor;
+
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        crate::scheduler::cleanup::run(
+            crate::scheduler::cleanup::standalone_pool(),
+            socket.into_close(),
+        )
+        .await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        crate::scheduler::cleanup::submit_only(
+            crate::scheduler::cleanup::standalone_pool(),
+            socket.into_close(),
+        )
+        .await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        let socket = socket.retain();
+        crate::scheduler::cleanup::run(crate::scheduler::cleanup::standalone_pool(), move || {
+            socket.get_ref().shutdown(how)
+        })
+        .await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        let socket = socket.retain();
+        crate::scheduler::cleanup::submit_only(
+            crate::scheduler::cleanup::standalone_pool(),
+            move || socket.get_ref().shutdown(how),
+        )
+        .await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
@@ -253,13 +295,13 @@ impl FileAllocate for Selector {
 impl File for Selector {
     type File = FileDescriptor;
     async fn file_close(&self, file: Self::File) -> io::Result<()> {
-        crate::scheduler::file_close::close(crate::scheduler::file_close::standalone_pool(), file)
+        crate::scheduler::file_close::close(crate::scheduler::cleanup::standalone_pool(), file)
             .await
     }
 
     async fn file_close_submit(&self, file: Self::File) -> io::Result<()> {
         crate::scheduler::file_close::submit_only(
-            crate::scheduler::file_close::standalone_pool(),
+            crate::scheduler::cleanup::standalone_pool(),
             file,
         )
         .await
diff --git a/crates/executor/src/scheduler/selector/readiness/socket_descriptor.rs b/crates/executor/src/scheduler/selector/readiness/socket_descriptor.rs
index 956d0d2..816f114 100644
--- a/crates/executor/src/scheduler/selector/readiness/socket_descriptor.rs
+++ b/crates/executor/src/scheduler/selector/readiness/socket_descriptor.rs
@@ -2,17 +2,11 @@
 // Copyright, 2026, by Samuel Williams.
 
 use async_io::Async;
-use std::{io, net::SocketAddr};
-
-#[cfg(all(feature = "io-uring", target_os = "linux"))]
-use std::sync::Arc;
+use std::{io, net::SocketAddr, sync::Arc};
 
 /// An owned nonblocking IP socket with one persistent native readiness registration. Unconnected, listening, connected, and datagram sockets use the same resource type. It is deliberately not `Clone`; backend requests privately retain ownership while in flight.
 pub struct SocketDescriptor {
-    #[cfg(all(feature = "io-uring", target_os = "linux"))]
     pub(crate) inner: Arc<Async<socket2::Socket>>,
-    #[cfg(not(all(feature = "io-uring", target_os = "linux")))]
-    pub(crate) inner: Async<socket2::Socket>,
 }
 
 impl SocketDescriptor {
@@ -25,16 +19,25 @@ impl SocketDescriptor {
         register: impl FnOnce(socket2::Socket) -> io::Result<Async<socket2::Socket>>,
     ) -> io::Result<Self> {
         let inner = register(socket)?;
-        #[cfg(all(feature = "io-uring", target_os = "linux"))]
         let inner = Arc::new(inner);
         Ok(Self { inner })
     }
-    #[cfg(all(feature = "io-uring", target_os = "linux"))]
     pub(crate) fn retain(&self) -> Self {
         Self {
             inner: Arc::clone(&self.inner),
         }
     }
+    pub(crate) fn into_close(self) -> impl FnOnce() -> io::Result<()> + Send + 'static {
+        // Establish exclusivity before admission, then let the cleanup worker deregister and close the socket together.
+        let inner = Arc::try_unwrap(self.inner)
+            .expect("cannot close a socket with outstanding operations; await them first");
+        move || {
+            inner
+                .into_inner()
+                .and_then(crate::scheduler::socket_io::close)
+        }
+    }
+
     pub(crate) fn get_ref(&self) -> &socket2::Socket {
         self.inner.get_ref()
     }
@@ -54,6 +57,20 @@ impl SocketDescriptor {
 mod tests {
     use super::*;
 
+    #[test]
+    fn explicit_close_rejects_retained_socket_ownership() {
+        let descriptor = SocketDescriptor::new(
+            socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap(),
+        )
+        .unwrap();
+        let retained = descriptor.retain();
+        let result =
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| descriptor.into_close()));
+        assert!(result.is_err());
+        assert!(retained.get_ref().set_nonblocking(true).is_ok());
+        retained.into_close()().unwrap();
+    }
+
     #[test]
     fn registration_errors_are_returned() {
         let socket =
diff --git a/crates/executor/src/scheduler/socket.rs b/crates/executor/src/scheduler/socket.rs
index dcb50ef..4123bf1 100644
--- a/crates/executor/src/scheduler/socket.rs
+++ b/crates/executor/src/scheduler/socket.rs
@@ -5,7 +5,11 @@ use crate::{
     BufferResult, Cancellation, Interest, IoBuf, IoBufMut, OperationResult, SocketDomain,
     SocketProtocol, SocketType,
 };
-use std::{future::Future, io, net::SocketAddr};
+use std::{
+    future::Future,
+    io,
+    net::{Shutdown, SocketAddr},
+};
 
 /// IP socket operations through the chosen scheduler. Each backend supplies one descriptor type for unconnected, listening, and connected sockets. Operations have the same state requirements as their system calls; invalid transitions report OS errors.
 ///
@@ -15,6 +19,31 @@ use std::{future::Future, io, net::SocketAddr};
 pub trait Socket: Send + Sync {
     type Socket: Send + Sync;
 
+    /// Consume and close an exclusively owned socket, awaiting bounded admission and OS completion. Finish all other operations first: close never cancels or joins them, and retained backend use may panic. Dropping an unpolled close releases the owned descriptor through its ordinary destructor.
+    fn socket_close(&self, socket: Self::Socket) -> impl Future<Output = io::Result<()>> + Send;
+
+    /// Consume the socket and await bounded cleanup admission only. Accepted close jobs survive waiter abandonment and are drained during graceful scheduler shutdown. Execution errors are discarded; use `socket_close` to observe them. The same exclusive-ownership requirement applies.
+    fn socket_close_submit(
+        &self,
+        socket: Self::Socket,
+    ) -> impl Future<Output = io::Result<()>> + Send;
+
+    /// Shut down reads, writes, or both, awaiting bounded admission and OS completion. The descriptor remains owned by the caller. Finish preceding writes before `Shutdown::Write`; reads may continue. Success confirms local shutdown, not peer receipt. Repeated shutdown and later I/O preserve platform-specific results.
+    ///
+    /// This lazy cleanup operation has no cancellation input. After admission it retains the socket and executes even if its waiter is dropped. Await its completion before explicitly closing the socket.
+    fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: Shutdown,
+    ) -> impl Future<Output = io::Result<()>> + Send;
+
+    /// Request shutdown and await bounded admission only. Acceptance does not establish completion or ordering with later operations, including close. The backend retains the socket until execution finishes; graceful scheduler shutdown drains accepted requests. Use `socket_shutdown` when subsequent work requires completion or errors.
+    fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: Shutdown,
+    ) -> impl Future<Output = io::Result<()>> + Send;
+
     /// Open a nonblocking socket without binding, listening, or connecting. `None` selects the OS default protocol for the requested domain and type.
     fn socket_open(
         &self,
diff --git a/crates/executor/src/scheduler/socket_io.rs b/crates/executor/src/scheduler/socket_io.rs
index cf16b1f..ed4fb7c 100644
--- a/crates/executor/src/scheduler/socket_io.rs
+++ b/crates/executor/src/scheduler/socket_io.rs
@@ -8,6 +8,51 @@ use socket2::{Domain, Protocol, Socket, Type};
 use std::future::Future;
 use std::{io, marker::PhantomData, net::SocketAddr, slice};
 
+// Called only on the cleanup pool after exclusive ownership and deregistration. Close is never retried on Unix: the descriptor may already have been released on error.
+pub(crate) fn close(socket: Socket) -> io::Result<()> {
+    #[cfg(unix)]
+    let result = {
+        use std::os::fd::IntoRawFd;
+        // SAFETY: The socket is exclusively owned and deregistered, and transferring the raw descriptor prevents a second close in Drop.
+        unsafe { libc::close(socket.into_raw_fd()) }
+    };
+    #[cfg(unix)]
+    return close_result(result);
+
+    #[cfg(windows)]
+    close_with(socket, |socket| socket.set_nonblocking(false))
+}
+
+#[cfg(windows)]
+fn close_with(
+    socket: Socket,
+    set_blocking: impl FnOnce(&Socket) -> io::Result<()>,
+) -> io::Result<()> {
+    use std::os::windows::io::IntoRawSocket;
+    // Winsock cannot close a nonblocking socket with positive linger reliably. The cleanup worker may block, and no operation or readiness registration still uses this socket.
+    set_blocking(&socket)?;
+    // SAFETY: Ownership is transferred out of Socket; no other backend operation retains this socket, and Drop must not close it again.
+    let result = unsafe {
+        windows_sys::Win32::Networking::WinSock::closesocket(socket.into_raw_socket() as _)
+    };
+    close_result(result)
+}
+
+fn close_result(result: i32) -> io::Result<()> {
+    if result == 0 {
+        Ok(())
+    } else {
+        #[cfg(unix)]
+        let error = io::Error::last_os_error();
+        #[cfg(windows)]
+        // SAFETY: WSAGetLastError reads the calling thread's most recent Winsock error.
+        let error = io::Error::from_raw_os_error(unsafe {
+            windows_sys::Win32::Networking::WinSock::WSAGetLastError()
+        });
+        Err(error)
+    }
+}
+
 pub(super) struct SendBuffer<'buffer> {
     pointer: *const u8,
     length: usize,
@@ -145,6 +190,74 @@ fn connected_with(
 mod tests {
     use super::*;
 
+    #[test]
+    fn explicit_close_consumes_socket_and_preserves_error_results() {
+        close(Socket::new(Domain::IPV4, Type::STREAM, None).unwrap()).unwrap();
+        assert!(close_result(-1).is_err());
+    }
+
+    #[test]
+    fn explicit_close_preserves_positive_linger_and_queued_data() {
+        use std::{io::Read, net::TcpListener, thread, time::Duration};
+
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP)).unwrap();
+        socket
+            .connect(&listener.local_addr().unwrap().into())
+            .unwrap();
+        let (mut peer, _) = listener.accept().unwrap();
+        peer.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+        socket2::SockRef::from(&peer)
+            .set_recv_buffer_size(8192)
+            .unwrap();
+        socket.set_send_buffer_size(4096).unwrap();
+        socket.set_linger(Some(Duration::from_secs(5))).unwrap();
+        socket.set_nonblocking(true).unwrap();
+
+        // Fill the send queue before the peer starts reading. Bound the payload as socket buffer limits vary by platform.
+        let mut sent = 0;
+        while sent < 1024 * 1024 {
+            let result = socket.send(&[0x5a; 4096]);
+            if matches!(&result, Err(error) if error.kind() == io::ErrorKind::WouldBlock) {
+                break;
+            }
+            let length = result.unwrap();
+            assert_ne!(length, 0);
+            sent += length;
+        }
+        assert!(sent > 0);
+        let reader = thread::spawn(move || {
+            let mut received = Vec::new();
+            peer.read_to_end(&mut received).unwrap();
+            received
+        });
+
+        close(socket).unwrap();
+        assert_eq!(reader.join().unwrap(), vec![0x5a; sent]);
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn close_configuration_errors_are_returned_and_ownership_is_released() {
+        use std::{
+            io::Read,
+            net::{TcpListener, TcpStream},
+            time::Duration,
+        };
+
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let socket = Socket::from(TcpStream::connect(listener.local_addr().unwrap()).unwrap());
+        let (mut peer, _) = listener.accept().unwrap();
+        peer.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+        socket.set_nonblocking(true).unwrap();
+
+        let error =
+            close_with(socket, |_| Err(io::ErrorKind::PermissionDenied.into())).unwrap_err();
+
+        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
+        assert_eq!(peer.read(&mut [0]).unwrap(), 0);
+    }
+
     #[test]
     fn socket_creation_errors_are_returned() {
         let error = io::Error::from(io::ErrorKind::PermissionDenied);
diff --git a/crates/executor/src/scheduler/socketry/operations.rs b/crates/executor/src/scheduler/socketry/operations.rs
index bcc783a..cc12605 100644
--- a/crates/executor/src/scheduler/socketry/operations.rs
+++ b/crates/executor/src/scheduler/socketry/operations.rs
@@ -89,6 +89,56 @@ impl SocketRegistration for SchedulerHandle {
 impl Socket for SchedulerHandle {
     type Socket = <DefaultSelector as Socket>::Socket;
 
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        check_open(
+            self.shared
+                .closed
+                .load(std::sync::atomic::Ordering::Acquire),
+        )?;
+        crate::scheduler::cleanup::run(&self.shared.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        check_open(
+            self.shared
+                .closed
+                .load(std::sync::atomic::Ordering::Acquire),
+        )?;
+        crate::scheduler::cleanup::submit_only(&self.shared.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        check_open(
+            self.shared
+                .closed
+                .load(std::sync::atomic::Ordering::Acquire),
+        )?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::run(&self.shared.cleanup, move || socket.get_ref().shutdown(how))
+            .await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        check_open(
+            self.shared
+                .closed
+                .load(std::sync::atomic::Ordering::Acquire),
+        )?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::submit_only(&self.shared.cleanup, move || {
+            socket.get_ref().shutdown(how)
+        })
+        .await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
@@ -332,6 +382,30 @@ impl SocketRegistration for Scheduler {
 impl Socket for Scheduler {
     type Socket = <SchedulerHandle as Socket>::Socket;
 
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        self.handle.socket_close(socket).await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        self.handle.socket_close_submit(socket).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.handle.socket_shutdown(socket, how).await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.handle.socket_shutdown_submit(socket, how).await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
diff --git a/crates/executor/src/scheduler/socketry/operations/tests.rs b/crates/executor/src/scheduler/socketry/operations/tests.rs
index 1215e21..b4e3051 100644
--- a/crates/executor/src/scheduler/socketry/operations/tests.rs
+++ b/crates/executor/src/scheduler/socketry/operations/tests.rs
@@ -215,3 +215,33 @@ fn scheduler_and_handle_forward_network_file_and_clock_operations() {
         scheduler.sleep(Duration::ZERO, &NEVER).await.unwrap();
     });
 }
+
+#[test]
+fn socket_cleanup_preserves_admitted_jobs_and_bounds_admission() {
+    for mode in 0..6 {
+        let capacity = if mode == 4 { 1 } else { 2 };
+        let scheduler = Scheduler::with_workers_and_blocking(
+            1,
+            crate::BlockingOptions {
+                threads: 1,
+                capacity,
+            },
+        )
+        .unwrap();
+        let handle = scheduler.handle();
+        scheduler.block_on(async {
+            let pool = &handle.shared.cleanup;
+            match mode {
+                0..=3 => {
+                    crate::scheduler::cleanup::tests::accepted_socket_jobs(&handle, pool, mode)
+                        .await
+                }
+                4 => {
+                    crate::scheduler::cleanup::tests::saturated_socket_admission(&handle, pool)
+                        .await
+                }
+                _ => crate::scheduler::cleanup::tests::closed_socket_admission(&handle, pool).await,
+            }
+        });
+    }
+}
diff --git a/crates/executor/src/scheduler/tokio.rs b/crates/executor/src/scheduler/tokio.rs
index 5812322..9f16929 100644
--- a/crates/executor/src/scheduler/tokio.rs
+++ b/crates/executor/src/scheduler/tokio.rs
@@ -539,6 +539,40 @@ impl SocketRegistration for SchedulerHandle {
 impl SocketOperations for SchedulerHandle {
     type Socket = SocketDescriptor;
 
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        self.check_runtime(socket.runtime())?;
+        crate::scheduler::cleanup::run(&self.shared.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        self.check_runtime(socket.runtime())?;
+        crate::scheduler::cleanup::submit_only(&self.shared.cleanup, socket.into_close()).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.check_runtime(socket.runtime())?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::run(&self.shared.cleanup, move || socket.get_ref().shutdown(how))
+            .await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.check_runtime(socket.runtime())?;
+        let socket = socket.retain();
+        crate::scheduler::cleanup::submit_only(&self.shared.cleanup, move || {
+            socket.get_ref().shutdown(how)
+        })
+        .await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
@@ -1009,6 +1043,30 @@ impl SocketRegistration for Scheduler {
 impl SocketOperations for Scheduler {
     type Socket = <SchedulerHandle as SocketOperations>::Socket;
 
+    async fn socket_close(&self, socket: Self::Socket) -> io::Result<()> {
+        self.handle.socket_close(socket).await
+    }
+
+    async fn socket_close_submit(&self, socket: Self::Socket) -> io::Result<()> {
+        self.handle.socket_close_submit(socket).await
+    }
+
+    async fn socket_shutdown(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.handle.socket_shutdown(socket, how).await
+    }
+
+    async fn socket_shutdown_submit(
+        &self,
+        socket: &Self::Socket,
+        how: std::net::Shutdown,
+    ) -> io::Result<()> {
+        self.handle.socket_shutdown_submit(socket, how).await
+    }
+
     async fn socket_open(
         &self,
         domain: crate::SocketDomain,
diff --git a/crates/executor/src/scheduler/tokio/socket_descriptor.rs b/crates/executor/src/scheduler/tokio/socket_descriptor.rs
index a66718e..47ee705 100644
--- a/crates/executor/src/scheduler/tokio/socket_descriptor.rs
+++ b/crates/executor/src/scheduler/tokio/socket_descriptor.rs
@@ -2,7 +2,7 @@
 // Copyright, 2026, by Samuel Williams.
 
 use crate::Interest;
-use std::{io, net::SocketAddr};
+use std::{io, net::SocketAddr, sync::Arc};
 use tokio::runtime::{Handle, Id};
 
 #[cfg(unix)]
@@ -13,9 +13,9 @@ type RegisteredSocket = async_io::Async<socket2::Socket>;
 
 /// A nonblocking socket registered with the adapter's Tokio runtime.
 ///
-/// A descriptor can represent an unconnected, bound, listening, connected, or datagram socket. It is deliberately not `Clone`; operation futures borrow it for their duration.
+/// A descriptor can represent an unconnected, bound, listening, connected, or datagram socket. It is deliberately not `Clone`; operation futures borrow it, and accepted shutdown jobs privately retain its registration until completion. Explicit close consumes exclusive ownership.
 pub struct SocketDescriptor {
-    inner: RegisteredSocket,
+    inner: Arc<RegisteredSocket>,
     runtime: Id,
 }
 
@@ -51,11 +51,35 @@ impl SocketDescriptor {
         let inner = register(runtime, socket)?;
 
         Ok(Self {
-            inner,
+            inner: Arc::new(inner),
             runtime: runtime.id(),
         })
     }
 
+    pub(super) fn retain(&self) -> Self {
+        Self {
+            inner: Arc::clone(&self.inner),
+            runtime: self.runtime,
+        }
+    }
+
+    pub(super) fn into_close(self) -> impl FnOnce() -> io::Result<()> + Send + 'static {
+        let inner = Arc::try_unwrap(self.inner)
+            .expect("cannot close a socket with outstanding operations; await them first");
+        move || {
+            #[cfg(unix)]
+            {
+                crate::scheduler::socket_io::close(inner.into_inner())
+            }
+            #[cfg(windows)]
+            {
+                inner
+                    .into_inner()
+                    .and_then(crate::scheduler::socket_io::close)
+            }
+        }
+    }
+
     pub(super) fn runtime(&self) -> Id {
         self.runtime
     }
@@ -159,6 +183,18 @@ mod tests {
         socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None).unwrap()
     }
 
+    #[test]
+    fn explicit_close_rejects_retained_socket_ownership() {
+        let runtime = runtime();
+        let descriptor = SocketDescriptor::new(runtime.handle(), socket()).unwrap();
+        let retained = descriptor.retain();
+        let result =
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| descriptor.into_close()));
+        assert!(result.is_err());
+        assert!(retained.get_ref().set_nonblocking(true).is_ok());
+        retained.into_close()().unwrap();
+    }
+
     #[test]
     fn nonblocking_configuration_errors_are_returned() {
         let runtime = runtime();
diff --git a/crates/executor/src/scheduler/tokio/tests.rs b/crates/executor/src/scheduler/tokio/tests.rs
index f8d20b6..609b4ed 100644
--- a/crates/executor/src/scheduler/tokio/tests.rs
+++ b/crates/executor/src/scheduler/tokio/tests.rs
@@ -681,3 +681,34 @@ fn cancelling_queued_blocking_work_preserves_buffers_cursor_and_prevents_creatio
     assert_eq!(file.as_ref().metadata().unwrap().len(), 0);
     runtime.block_on(scheduler.shutdown());
 }
+
+#[test]
+fn socket_cleanup_preserves_admitted_jobs_and_bounds_admission() {
+    for mode in 0..6 {
+        let capacity = if mode == 4 { 1 } else { 2 };
+        let runtime = runtime();
+        let scheduler = Scheduler::with_blocking(
+            runtime.handle().clone(),
+            crate::BlockingOptions {
+                threads: 1,
+                capacity,
+            },
+        )
+        .unwrap();
+        let handle = scheduler.handle();
+        runtime.block_on(async {
+            let pool = &handle.shared.cleanup;
+            match mode {
+                0..=3 => {
+                    crate::scheduler::cleanup::tests::accepted_socket_jobs(&handle, pool, mode)
+                        .await
+                }
+                4 => {
+                    crate::scheduler::cleanup::tests::saturated_socket_admission(&handle, pool)
+                        .await
+                }
+                _ => crate::scheduler::cleanup::tests::closed_socket_admission(&handle, pool).await,
+            }
+        });
+    }
+}
diff --git a/crates/executor/tests/contracts.rs b/crates/executor/tests/contracts.rs
index d6629aa..9cad2f3 100644
--- a/crates/executor/tests/contracts.rs
+++ b/crates/executor/tests/contracts.rs
@@ -31,7 +31,11 @@ mod file_allocate;
 #[path = "../examples/interface_probe/tests/file_close.rs"]
 mod file_close;
 
+#[path = "../examples/interface_probe/tests/socket_lifecycle.rs"]
+mod socket_lifecycle;
+
 async fn resources<S: File + Socket>(scheduler: &S, root: &Path) {
+    socket_lifecycle::check(scheduler).await;
     file_close::check(scheduler, root).await;
     stateful_file::check(scheduler, root).await;
     file_sync::check(scheduler, root).await;
@@ -504,6 +508,7 @@ fn native_resources() {
         let fourth = tempfile::tempdir().unwrap();
         let handle = scheduler.handle();
         scheduler.block_on(file_close::check(handle.selector().unwrap(), fourth.path()));
+        scheduler.block_on(socket_lifecycle::check(handle.selector().unwrap()));
     }
 }
 
diff --git a/crates/executor/tests/socket_lifecycle.rs b/crates/executor/tests/socket_lifecycle.rs
new file mode 100644
index 0000000..58a9ca0
--- /dev/null
+++ b/crates/executor/tests/socket_lifecycle.rs
@@ -0,0 +1,93 @@
+// Released under the MIT License.
+// Copyright, 2026, by Samuel Williams.
+
+#![cfg(any(feature = "native", feature = "tokio"))]
+
+use socketry_executor::{Cancellation, Socket, SocketDomain, SocketType};
+use std::{io, net::Shutdown};
+
+async fn sockets<S: Socket>(scheduler: &S) -> [S::Socket; 3] {
+    async fn open<S: Socket>(scheduler: &S) -> S::Socket {
+        scheduler
+            .socket_open(
+                SocketDomain::Ipv4,
+                SocketType::Stream,
+                None,
+                &Cancellation::never(),
+            )
+            .await
+            .unwrap()
+    }
+    [
+        open(scheduler).await,
+        open(scheduler).await,
+        open(scheduler).await,
+    ]
+}
+
+async fn rejected<S: Socket>(
+    scheduler: &S,
+    [first, second, third]: [S::Socket; 3],
+    expected: io::ErrorKind,
+) {
+    assert_eq!(
+        scheduler
+            .socket_shutdown(&first, Shutdown::Write)
+            .await
+            .unwrap_err()
+            .kind(),
+        expected
+    );
+    assert_eq!(
+        scheduler
+            .socket_shutdown_submit(&first, Shutdown::Write)
+            .await
+            .unwrap_err()
+            .kind(),
+        expected
+    );
+    assert_eq!(
+        scheduler.socket_close(second).await.unwrap_err().kind(),
+        expected
+    );
+    assert_eq!(
+        scheduler
+            .socket_close_submit(third)
+            .await
+            .unwrap_err()
+            .kind(),
+        expected
+    );
+}
+
+#[cfg(feature = "native")]
+#[test]
+fn closed_native_scheduler_rejects_socket_lifecycle_operations() {
+    let scheduler = socketry_executor::Scheduler::with_workers(1).unwrap();
+    let handle = scheduler.handle();
+    let sockets = scheduler.block_on(sockets(&scheduler));
+    scheduler.shutdown();
+    futures_lite::future::block_on(rejected(&handle, sockets, io::ErrorKind::BrokenPipe));
+}
+
+#[cfg(feature = "tokio")]
+#[test]
+fn tokio_lifecycle_rejects_closed_schedulers_and_foreign_resources() {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let scheduler = socketry_executor::scheduler::tokio::Scheduler::new(runtime.handle().clone());
+    let other_runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let other = socketry_executor::scheduler::tokio::Scheduler::new(other_runtime.handle().clone());
+    let foreign = runtime.block_on(sockets(&scheduler));
+    other_runtime.block_on(rejected(&other, foreign, io::ErrorKind::InvalidInput));
+    let sockets = runtime.block_on(sockets(&scheduler));
+    let handle = scheduler.handle();
+    runtime.block_on(scheduler.shutdown());
+    runtime.block_on(rejected(&handle, sockets, io::ErrorKind::BrokenPipe));
+    other_runtime.block_on(other.shutdown());
+}
diff --git a/readme.md b/readme.md
index 8775c5d..33b267a 100644
--- a/readme.md
+++ b/readme.md
@@ -80,6 +80,8 @@ Generic code can accept `Socket`, `File`, `Clock`, and `Spawn` capabilities. Soc
 
 `file_close` waits for OS completion; `file_close_submit` waits only for bounded admission. Both consume an owned file handle and require prior operations to have finished. The [commented file-close example](crates/executor/examples/file_close.rs) demonstrates requesting cancellation, awaiting completion, and then closing safely.
 
+`socket_shutdown(&socket, how)` supports directional shutdown while retaining the descriptor. `socket_close(socket)` consumes it after outstanding work has finished. Both have admission-only `_submit` forms. The [commented half-close example](crates/executor/examples/socket_shutdown.rs) sends a request, shuts down writes, receives the response, then closes.
+
 `FileAllocate` adds optional range allocation with `AllocationMode::Extend` or `KeepSize`. Native and Tokio implementations currently provide it on Linux; unsupported targets omit the trait implementation. See the [executor I/O documentation](crates/executor/readme.md#io-and-selectors) for allocation and filesystem guarantees.
 
 `File` is the scheduler capability trait; `File::File` is its associated resource type. Generic code can create and use files without importing OS resource types:
diff --git a/releases.md b/releases.md
index 4f7cba8..8fc0f50 100644
--- a/releases.md
+++ b/releases.md
@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Add required `Socket::socket_shutdown`, `socket_shutdown_submit`, `socket_close`, and `socket_close_submit` across native, Tokio, io_uring, and simulated sockets. Directional shutdown borrows the descriptor; close consumes exclusive ownership. Accepted cleanup shares bounded admission and graceful draining with file close. Add a commented half-close example and conformance/lifetime tests. External `Socket` implementations must supply all four operations.
+
 - Add required `File::file_close` and `file_close_submit`, owned non-`Clone` native/Tokio `FileDescriptor` resources, bounded cleanup admission, and graceful cleanup draining. Finish outstanding operations before explicit close; abandoned backend use causes a panic rather than implicit cancellation or joining. Add a commented cancellation/close example and release io_uring file references before completion wakeups. External `File` implementations must add both methods; concrete callers using `Arc<std::fs::File>` must migrate to `FileDescriptor`.
 
 - Add optional `FileAllocate` with `AllocationMode::Extend`/`KeepSize`, cancellation, and cursor preservation. Linux native and Tokio backends reserve space with `fallocate`; io\_uring uses native requests when supported. Other targets omit the capability, and filesystem errors remain observable.