diff --git a/issues/a-close-of-one-handle-ends-every-rings-poll-on-its-object.md b/issues/a-close-of-one-handle-ends-every-rings-poll-on-its-object.md
index 4d0aec608cb..910446a18c6 100644
--- a/issues/a-close-of-one-handle-ends-every-rings-poll-on-its-object.md
+++ b/issues/a-close-of-one-handle-ends-every-rings-poll-on-its-object.md
@@ -9,12 +9,14 @@ opened: 2026-10-02
 `ops::close` answers every poll on a watch its object ends with `-NotFound`
 (`Watch::cancel_polls`), in every ring, when any one handle to a pipe's read
 end or an acceptor closes. A sibling handle from `dup` keeps the object open,
-and its polls end all the same. `toyos::poller`'s `drain` hands the token of a
-negative completion to its caller as it does a ready one's, so a reader that
-takes that token for bytes and reads blocking parks on an object that is open
-and empty. No caller in the tree reaches it: fsd's acceptors are endowed and
+and its polls end all the same. `toyos::poller`'s `wait` hands the token of a
+negative completion to its caller as it does a ready one's and drops the
+kernel's word for it, so a reader that takes that token for bytes and reads
+blocking parks on an object that is open and empty. `wait_answers` hands the
+word beside the token. No caller in the tree reaches it: fsd's acceptors are endowed and
 never duplicated. `inbox_cancel_wakes` stages the close.
 
-**Exit**: a poll ends only when the last handle to its source closes, or a
-completion's result reaches `Poller`'s caller; a test closes a duplicate under
-a watch and reads what the wait hands back.
+**Exit**: a poll ends only when the last handle to its source closes, or
+`wait`, the form that drops a completion's result, is gone and every caller of
+`Poller` reads it through `wait_answers`; a test closes a duplicate under a
+watch and reads what the wait hands back.
diff --git a/issues/a-shutdown-of-the-sending-half-drops-what-the-send-pipe-still-holds.md b/issues/a-shutdown-of-the-sending-half-drops-what-the-send-pipe-still-holds.md
new file mode 100644
index 00000000000..e9f641083a1
--- /dev/null
+++ b/issues/a-shutdown-of-the-sending-half-drops-what-the-send-pipe-still-holds.md
@@ -0,0 +1,29 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# A shutdown of the sending half drops what the send pipe still holds
+
+`handle_tcp_shutdown` (`userland/netstack/src/main.rs`) calls `socket.close()`
+on the pass that reads the request. A client's bytes reach netstack through
+its send pipe and its request through its connection, and nothing orders the
+two: bytes the client wrote before it asked may still be in the pipe.
+`bridge_piped` reads the pipe only while `send_room` holds, which is false
+from `FIN-WAIT-1` on, so those bytes are never sent and the peer reads a
+stream that ends short with a clean FIN.
+
+Dropping the write end instead is the path that works: the bridge reads the
+pipe to its end and closes the socket after the last byte.
+
+Read from the code, not measured. `netstack_socket_churn`
+(`tests/toyos-rust-tests/src/bin/netstack_socket_churn.rs`) writes into the
+pipe after the shutdown, which is the client's own error and not this.
+
+**Exit condition**: a shutdown of the sending half closes the socket after
+the bytes the pipe held when the request was read, and a guest test on
+`tests/netcase` whose client writes, shuts down at once and reads the peer's
+echo sees every byte it wrote.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/a-stream-its-peer-reset-refuses-the-option-requests-a-host-answers.md b/issues/a-stream-its-peer-reset-refuses-the-option-requests-a-host-answers.md
new file mode 100644
index 00000000000..fd1f1df5948
--- /dev/null
+++ b/issues/a-stream-its-peer-reset-refuses-the-option-requests-a-host-answers.md
@@ -0,0 +1,35 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# A stream its peer reset refuses the option requests a host answers
+
+netstack lets a piped connection's socket and id go once the wire is finished
+and it has closed both of its pipe ends (`bridge_piped`,
+`userland/netstack/src/main.rs`). A peer's reset does both at once, so a
+client that still holds the stream names an id netstack no longer has, and
+`handle_tcp_set_option` and `handle_tcp_get_option` answer it
+`ERR_NOT_CONNECTED`. Before the socket left with its bridge both answered from
+the kept socket.
+
+What a host answers `std::net::TcpStream` on a stream whose peer reset it,
+after the read that reported the reset:
+
+| | `set_nodelay(true)` | `nodelay()` |
+|---|---|---|
+| macOS (Darwin 27.0.0), measured | `Err(InvalidInput)`, `EINVAL` | `Ok(false)` |
+| Linux, by the review of the change that made this, not measured | `Ok(())` | `Ok` |
+| ToyOS, read from the code | `Err(NotConnected)` | `Err(NotConnected)` |
+
+So the write is refused as macOS refuses it, under another kind, and the read
+is refused where both hosts answer. Keeping the id until the client's close
+request is not the fix: a client that dies sends none, and netstack, having
+closed its pipe ends, has nothing left that says the client is gone.
+
+**Exit condition**: a guest test on `tests/netcase` whose peer resets a stream
+the client still holds, and `nodelay()` on it answers `Ok`, with
+`netstack_socket_churn` still green.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/logkeeper-hears-a-reader-leave-only-at-its-next-write.md b/issues/logkeeper-hears-a-reader-leave-only-at-its-next-write.md
new file mode 100644
index 00000000000..713c554c6af
--- /dev/null
+++ b/issues/logkeeper-hears-a-reader-leave-only-at-its-next-write.md
@@ -0,0 +1,20 @@
+---
+status: open
+kind: finding
+opened: 2026-10-08
+---
+
+# logkeeper hears a reader leave only at its next write
+
+A log reader's thread (`userland/logkeeper/src/serve.rs`, `feed`) waits on
+`shared.grew` once it has caught up, and learns its reader is gone only when a
+write to the sink fails. A reader that leaves while the log is quiet keeps its
+thread and its place in `MAX_NETWORK_READERS` or `MAX_LOCAL_READERS` until the
+next record lands.
+
+Read from the code, not measured.
+
+**Exit condition**: a reader's end is what wakes its thread, or the header of
+`serve.rs` says why the next record is soon enough.
+
+**Owner**: whoever holds `issues/redesign-the-log-subsystem.md`.
diff --git a/issues/netstack-cuts-a-departed-clients-unsent-tail-at-the-ceiling.md b/issues/netstack-cuts-a-departed-clients-unsent-tail-at-the-ceiling.md
new file mode 100644
index 00000000000..5e125bf6c5f
--- /dev/null
+++ b/issues/netstack-cuts-a-departed-clients-unsent-tail-at-the-ceiling.md
@@ -0,0 +1,46 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# netstack cuts a departed client's unsent tail at the ceiling
+
+A piped connection whose client holds no end of a direction still open
+(`userland/netstack/src/main.rs`, `clientless`) has `OWNERLESS_LIFE`, 100
+seconds from the pass that found it so, to finish on the wire (`ownerless`).
+That is every client that is gone, and one that is not: a client that let go
+of its send end and still holds the read end of a receive pipe netstack closed
+at the peer's FIN is timed and cut the same way, alive. The bound is absolute: at the
+ceiling the socket is reset whatever it still owes its peer. A program that
+writes and exits is the ordinary case of a client that is gone, so when its
+peer takes longer than 100 seconds to acknowledge what was written, the peer's
+stream ends in a reset with the tail undelivered: whatever is left of the
+64 KiB send buffer, and whatever the client's send pipe still held, up to the
+pipe's 2 MiB.
+
+The cut is loud at both ends: netstack logs it and the peer reads a reset, not
+a short stream. Nobody is left to tell on the client's side.
+
+Hosts do otherwise for an orphaned connection with data still to send: it is
+cut on no progress or under orphan pressure and not on a clock from the close.
+From knowledge of Linux's `tcp_orphan_retries` and `tcp_max_orphans`, not
+measured here.
+
+The bound is absolute because the slot table has no per-client share
+(`issues/netstack-lookup-slots-have-no-per-client-share.md` records the same of
+the lookups): a peer that acknowledges a byte at a time would hold a slot of a
+client that is gone for as long as it liked, and
+`max_piped_connections` such peers hold every slot.
+
+Read from the code, not measured: the bench rows that reach the ceiling
+(`userland/netstack/src/listen/tests.rs`) cut a connection whose peer
+acknowledges nothing.
+
+**Exit condition**: a connection with no client whose peer keeps
+acknowledging new bytes is kept while it makes progress, under a bound on how
+many such connections one peer or one departed client may hold; and a bench
+row whose peer acknowledges one segment a second past the ceiling sees the
+whole tail and a FIN.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/netstack-datagram-sockets-and-listeners-have-no-bound.md b/issues/netstack-datagram-sockets-and-listeners-have-no-bound.md
new file mode 100644
index 00000000000..b0b895bac95
--- /dev/null
+++ b/issues/netstack-datagram-sockets-and-listeners-have-no-bound.md
@@ -0,0 +1,25 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# netstack's datagram sockets and listeners have no bound
+
+`max_piped_connections` (`userland/netstack/src/main.rs`) bounds piped TCP
+connections and pending connects, and `resolve::MAX_LOOKUPS` the lookups in
+flight. `handle_udp_bind` and `handle_tcp_bind_piped` check neither and
+nothing else: every bind adds a socket with two 64 KiB buffers and holds the
+client's 2 MiB pipes with it, two for a UDP socket and one for a listener. One holder of the `netstack`
+connector that binds in a loop takes netstack's memory, and every dynamic UDP
+port, from every other program. No bound is per client either:
+`issues/netstack-lookup-slots-have-no-per-client-share.md` records that for
+the two that exist.
+
+Read from the code, not measured.
+
+**Exit condition**: a bind past a stated bound is refused
+`ERR_RESOURCE_EXHAUSTED` with nothing made, and a test binds past it and sees
+the refusal and another client's bind answered.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/netstack-keeps-the-datagram-socket-of-a-client-that-sent-no-close.md b/issues/netstack-keeps-the-datagram-socket-of-a-client-that-sent-no-close.md
new file mode 100644
index 00000000000..ace4fdf6066
--- /dev/null
+++ b/issues/netstack-keeps-the-datagram-socket-of-a-client-that-sent-no-close.md
@@ -0,0 +1,36 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# netstack keeps the datagram socket of a client that sent no close
+
+A UDP socket is three things in netstack (`userland/netstack/src/main.rs`):
+the smoltcp socket with its two 64 KiB buffers, the id's entry in `sockets`,
+and the two pipe ends in `udp_pipes`, each keeping a 2 MiB kernel pipe alive.
+A close request removes all three. Nothing else looks at the pipes: a client
+that dies, or drops its ends without the request, leaves the socket and its
+bound port for the life of the process, and a restarted program that binds
+the same port is answered `ERR_ADDR_IN_USE`. The one other way out is
+`deliver_datagram`, which ends the socket when a receive that was already
+pending writes into a pipe with no reader.
+
+A listener's owner is heard, and only on a pass something else causes:
+`serve_piped_listeners` writes zero bytes to every notify pipe each pass and
+ends the listener the kernel answers `Gone` for, and no watch wakes a pass for
+it, so an idle netstack keeps a dead owner's port until other traffic arrives.
+
+A piped TCP connection is the shape to copy: `bridge_piped` ends it on what
+the kernel says of the client's pipe ends, and the watch on each of them is
+what wakes that pass.
+
+Read from the code, not measured.
+
+**Exit condition**: a datagram socket and a listener whose owner's pipe ends
+are closed leave the socket set and the table with no other traffic, and a
+guest test on `tests/netcase` that binds each, drops it without a close
+request, and reads `net.sockets.udp` and `net.sockets.listeners` back at what
+they were.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/netstack-keeps-the-socket-of-a-connection-whose-client-sent-no-close.md b/issues/netstack-keeps-the-socket-of-a-connection-whose-client-sent-no-close.md
deleted file mode 100644
index c67131a84d0..00000000000
--- a/issues/netstack-keeps-the-socket-of-a-connection-whose-client-sent-no-close.md
+++ /dev/null
@@ -1,41 +0,0 @@
----
-status: open
-kind: defect
-opened: 2026-10-08
----
-
-# netstack keeps the socket of a connection whose client sent no close
-
-A piped TCP stream has three things in netstack
-(`userland/netstack/src/main.rs`): the bridge in `piped_connections`, the
-smoltcp socket in the socket set with its two 64 KiB buffers, and the id's
-entry in `sockets`. A client's close request removes all three. When a
-connection ends any other way — the peer finishes or resets and the client
-drops its pipe ends, or the client dies — `bridge_piped` removes the bridge
-and nothing else. The socket and its entry stay for the life of the process.
-
-`max_piped_connections` counts bridges and pending connects, so each ended
-connection gives its slot back while its socket stays: the bound netstack
-states for what clients can make it hold does not cover this, and every
-client holding the `netstack` connector can grow it one connection at a time.
-`inspect` reports the entries in `net.sockets.tcp` and has no count that tells
-a kept one from a live one.
-
-**Measured** at `06bbc236d`, on `tests/netcase` under QEMU against a host
-server that ends each connection at once:
-`cargo test --test toyos-build -- netstack_socket_churn` exits 1 on
-
-    netstack_socket_churn: 4 connections ended and let go; netstack held 0 stream(s) before them and holds 4 after
-
-each of the four read after `net.piped.live` had returned to its reading
-before the first. The same run with the close request sent before the drop
-exits 0.
-
-**Exit condition**: a connection netstack lets go leaves no socket and no
-table entry, whoever ended it, and `netstack_socket_churn` is green.
-
-**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
-
-**Its test is deleted**: `8dd55d06a` took `netstack_socket_churn` out, with
-the `netstack` connector `tests/netcase`'s runner held for it, and
-`git revert 8dd55d06a` brings both back.
diff --git a/issues/netstack-spends-two-poller-slots-per-piped-connection.md b/issues/netstack-spends-two-poller-slots-per-piped-connection.md
index 23300c420f2..7c0a60c78d8 100644
--- a/issues/netstack-spends-two-poller-slots-per-piped-connection.md
+++ b/issues/netstack-spends-two-poller-slots-per-piped-connection.md
@@ -4,22 +4,27 @@ kind: defect
 opened: 2026-09-26
 ---
 
-# netd spends two poller slots per piped connection
+# netstack spends two poller slots per piped connection
 
-A piped connection registers its send pipe `READABLE` and, while it holds
-bytes its receive pipe refused, its receive pipe `WRITABLE` — both in one pass,
-because an ssh receiver sends `WINDOW_ADJUST` while its own receive is held.
-`MAX_PIPED_SLOTS` (`userland/netd/src/main.rs`) divides the batch one poller
-can carry (`Poller::MAX_HANDLES`, less the two fixed registrations and the
-pending connections) by `POLL_HANDLES_PER_PIPED = 2`, so the ceiling on live
-piped connections halved from 222 to 111.
+A piped connection registers both its pipes on every pass, for as long as
+netstack holds them: its send pipe `READABLE` while the socket has send room
+and `OTHER_END_GONE` until the kernel has answered that, and its receive pipe
+`OTHER_END_GONE` and, while it holds bytes the pipe refused, `WRITABLE`
+(`userland/netstack/src/main.rs`, the loop in `main`). `MAX_PIPED_SLOTS`
+divides the batch one poller can carry (`Poller::MAX_HANDLES`, less the fixed
+registrations, the pending connections and the lookups' clients) by
+`POLL_HANDLES_PER_PIPED = 2`, so the ceiling on live piped connections is half
+what one registration each would give.
 
-The memory budget (an eighth of memory at 4 MiB a connection) is the lower
-bound only below 3552 MiB (111 × 4 MiB × 8), so every machine with 4 GiB
-or more is held to 111.
+The memory budget (an eighth of memory at 4 MiB a connection) binds first only
+on a machine whose eighth holds fewer connections than that ceiling.
 
-Exit condition: one registration per connection. netd's side of a
-connection is one kernel `Connection` object, which `read_source` and
-`write_source` both resolve (`kernel/src/object/ops.rs`), so one `OP_WATCH`
-carries `READABLE | WRITABLE` (`kernel/src/inbox/mod.rs`, `process_watch`), and
-`POLL_HANDLES_PER_PIPED` is deleted.
+**Exit condition**: one registration per connection, and
+`POLL_HANDLES_PER_PIPED` deleted. One `OP_WATCH` on a joined `Connection`
+carries `READABLE | WRITABLE` for both its pipes, and is refused
+`OTHER_END_GONE`: `ops::pipe_end_watch` (`kernel/src/object/ops.rs`) answers a
+pipe end alone, and a connection has two other ends for the one bit. So the
+exit also needs the kernel to say of a connection which of its directions has
+no holder left.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/issues/netstack-watches-both-pipes-of-every-connection-on-every-pass.md b/issues/netstack-watches-both-pipes-of-every-connection-on-every-pass.md
new file mode 100644
index 00000000000..5b0421f65ce
--- /dev/null
+++ b/issues/netstack-watches-both-pipes-of-every-connection-on-every-pass.md
@@ -0,0 +1,32 @@
+---
+status: open
+kind: defect
+opened: 2026-10-08
+---
+
+# netstack watches both pipes of every connection on every pass
+
+Every pass of netstack's loop submits a watch for each pipe of each live piped
+connection, ready or not, changed or not (`userland/netstack/src/main.rs`, the
+loop in `main`): an idle connection submits two. A watch replaces its handle's
+earlier one, so each is a poll allocated and registered again, with the takes
+of the global pipe lock that `arm` makes to read the pipe and find its watches
+(`kernel/src/inbox/mod.rs`).
+
+A poll that asks `OTHER_END_GONE` sits on its own end's watch, which is the
+watch that end's readers or writers park on. So every write a client makes
+through the kernel posts the poll netstack keeps on that connection's send
+pipe, and every read the poll on its receive pipe: netstack's submitter is woken in the kernel for a
+look that finds the other end held and arms again. No completion is written and
+no pass runs, and the wake is paid per client read and write.
+
+Read from the code, not measured: nothing in the tree reads netstack's
+throughput or its passes per second with connections open, and a duration is
+the T14's to say.
+
+**Exit condition**: a pipe's watch is submitted only when its interest changes
+or the kernel has answered it, and a post that is no leaving wakes no
+`OTHER_END_GONE` poll; or a throughput reading on the T14, the same transfer
+with and without the watches, that shows the cost is nothing.
+
+**Owner**: whoever holds `issues/toyos-has-its-own-network-stack.md`.
diff --git a/kernel/pure/lib.rs b/kernel/pure/lib.rs
index 86f6f211fda..f3a288bec08 100644
--- a/kernel/pure/lib.rs
+++ b/kernel/pure/lib.rs
@@ -1,7 +1,8 @@
 //! What the kernel decides without touching the machine: the scheduler core
 //! ([`sched`]), the process and thread lifecycle ([`proclife`]), which PCID an
-//! address space is handed ([`pcid`]) and what type the range registers give a
-//! range ([`mtrr`]). The kernel binary links it; the host runs
+//! address space is handed ([`pcid`]), what type the range registers give a
+//! range ([`mtrr`]) and what a pipe's ends are told of each other ([`pipe`]).
+//! The kernel binary links it; the host runs
 //! its tests, because none of it reads a register, a clock or a kernel lock.
 
 #![no_std]
@@ -13,5 +14,6 @@ extern crate std;
 
 pub mod mtrr;
 pub mod pcid;
+pub mod pipe;
 pub mod proclife;
 pub mod sched;
diff --git a/kernel/pure/pipe.rs b/kernel/pure/pipe.rs
new file mode 100644
index 00000000000..76b2b39ac16
--- /dev/null
+++ b/kernel/pure/pipe.rs
@@ -0,0 +1,166 @@
+//! Who holds a pipe's two ends, and what a watch on one of them is told.
+//!
+//! A pipe end is held by counted references, one per object that names it: a
+//! handle duplicated or moved to another process is the same object or another
+//! reference, so an end is gone only when the last of them is. Every answer
+//! here is a function of the two counts and of what the ring holds, read under
+//! the kernel's pipe lock.
+
+#![forbid(unsafe_code)]
+
+/// One end of a pipe.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum End {
+    Read,
+    Write,
+}
+
+impl End {
+    pub fn other(self) -> Self {
+        match self {
+            Self::Read => Self::Write,
+            Self::Write => Self::Read,
+        }
+    }
+}
+
+/// What a reference's release left of its pipe.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+#[must_use = "a release that left an end unheld owes the other end a post"]
+pub enum Released {
+    /// Its end has another holder.
+    Held,
+    /// It was its end's last holder: the other end's watch is owed a post.
+    EndGone,
+    /// It was the pipe's last holder.
+    PipeGone,
+}
+
+/// How many references hold each end of one pipe.
+#[derive(Default)]
+pub struct Holders {
+    readers: u32,
+    writers: u32,
+}
+
+impl Holders {
+    pub const fn new() -> Self {
+        Self { readers: 0, writers: 0 }
+    }
+
+    fn count(&mut self, end: End) -> &mut u32 {
+        match end {
+            End::Read => &mut self.readers,
+            End::Write => &mut self.writers,
+        }
+    }
+
+    pub fn hold(&mut self, end: End) {
+        let count = self.count(end);
+        *count = count.checked_add(1).expect("pipe holder overflow");
+    }
+
+    pub fn release(&mut self, end: End) -> Released {
+        let count = self.count(end);
+        *count = count.checked_sub(1).expect("pipe holder underflow");
+        match (self.readers, self.writers) {
+            (0, 0) => Released::PipeGone,
+            _ if !self.held(end) => Released::EndGone,
+            _ => Released::Held,
+        }
+    }
+
+    pub fn held(&self, end: End) -> bool {
+        match end {
+            End::Read => self.readers > 0,
+            End::Write => self.writers > 0,
+        }
+    }
+
+    /// A read answers now: with bytes, or with the end of a stream no writer
+    /// can add to.
+    pub fn readable(&self, available: u32) -> bool {
+        available > 0 || !self.held(End::Write)
+    }
+
+    /// A write answers now: it takes bytes, or is refused for want of a reader.
+    pub fn writable(&self, space: u32) -> bool {
+        space > 0 || !self.held(End::Read)
+    }
+
+    /// The end opposite `end` has no holder, whatever the ring holds.
+    pub fn other_end_gone(&self, end: End) -> bool {
+        !self.held(end.other())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A pipe as `pipe::create` leaves it: one holder of each end.
+    fn made() -> Holders {
+        let mut pipe = Holders::new();
+        pipe.hold(End::Read);
+        pipe.hold(End::Write);
+        pipe
+    }
+
+    #[test]
+    fn an_end_whose_other_end_is_held_is_not_told_it_is_gone() {
+        let pipe = made();
+        assert!(!pipe.other_end_gone(End::Read));
+        assert!(!pipe.other_end_gone(End::Write));
+    }
+
+    /// **Whatever the ring holds.** `readable` answers for the bytes and says
+    /// nothing of the writer; `writable` for the room and nothing of the reader.
+    #[test]
+    fn an_end_dropped_with_the_ring_neither_empty_nor_full_is_gone() {
+        for (available, space) in [(0, 8), (3, 5), (8, 0)] {
+            let mut pipe = made();
+            assert_eq!(pipe.release(End::Write), Released::EndGone);
+            assert!(pipe.other_end_gone(End::Read), "the writer left with {available} byte(s) in the ring");
+            assert!(pipe.readable(available));
+            assert!(!pipe.other_end_gone(End::Write), "a reader that is held was told gone");
+
+            let mut pipe = made();
+            assert_eq!(pipe.release(End::Read), Released::EndGone);
+            assert!(pipe.other_end_gone(End::Write), "the reader left with room for {space} byte(s)");
+            assert!(pipe.writable(space));
+            assert!(!pipe.other_end_gone(End::Read), "a writer that is held was told gone");
+        }
+    }
+
+    /// The two older conditions do not stand in for it: each is ready with the
+    /// other end held.
+    #[test]
+    fn readable_and_writable_say_nothing_of_the_other_end() {
+        let pipe = made();
+        assert!(pipe.readable(1) && !pipe.other_end_gone(End::Read));
+        assert!(pipe.writable(1) && !pipe.other_end_gone(End::Write));
+        assert!(!pipe.readable(0) && !pipe.writable(0));
+    }
+
+    /// A handle duplicated, or moved to another process, is a second
+    /// reference: the end is gone with the last of them and not before.
+    #[test]
+    fn an_end_with_a_second_holder_outlives_the_first() {
+        for end in [End::Read, End::Write] {
+            let mut pipe = made();
+            pipe.hold(end);
+            assert_eq!(pipe.release(end), Released::Held);
+            assert!(!pipe.other_end_gone(end.other()), "{end:?} was told gone with a holder left");
+            assert_eq!(pipe.release(end), Released::EndGone);
+            assert!(pipe.other_end_gone(end.other()));
+        }
+    }
+
+    /// The pipe's last reference owes nobody a post: no end is left to watch.
+    #[test]
+    fn the_last_holder_of_the_pipe_ends_it() {
+        let mut pipe = made();
+        assert_eq!(pipe.release(End::Read), Released::EndGone);
+        assert_eq!(pipe.release(End::Write), Released::PipeGone);
+    }
+}
diff --git a/kernel/src/inbox/mod.rs b/kernel/src/inbox/mod.rs
index dc385159942..32310a66ae0 100644
--- a/kernel/src/inbox/mod.rs
+++ b/kernel/src/inbox/mod.rs
@@ -4,7 +4,7 @@
 //! into both kernel and userspace. `OP_WATCH` fires once; userspace re-submits to re-arm.
 //!
 //! **A watch is a poll registered on the watched object's own
-//! [`Watch`](crate::watch::Watch)**, one entry per direction it asked for, and
+//! [`Watch`](crate::watch::Watch)**, one entry per watch its interest names, and
 //! the object's post fires it. There is no table of sources here: what a
 //! handle watches is `ops::read_watch`/`ops::write_watch`'s answer, and the
 //! poll holds no reference to the object at all.
@@ -109,10 +109,11 @@ pub struct WatchFlags(u32);
 impl WatchFlags {
     pub const READABLE: Self = Self(toyos_abi::inbox::READABLE);
     pub const WRITABLE: Self = Self(toyos_abi::inbox::WRITABLE);
+    pub const OTHER_END_GONE: Self = Self(toyos_abi::inbox::OTHER_END_GONE);
     /// Every bit `toyos_abi::inbox` defines for `Submission::op_flags`;
     /// hand-copied and unchecked, for the reason `syscall/vm.rs`'s
     /// `MMAP_PROT_KNOWN` gives for all four of these masks.
-    const KNOWN: u32 = Self::READABLE.0 | Self::WRITABLE.0;
+    const KNOWN: u32 = Self::READABLE.0 | Self::WRITABLE.0 | Self::OTHER_END_GONE.0;
 
     /// A bit outside [`Self::KNOWN`] is an interest this kernel would register
     /// for neither direction, so the whole watch is refused rather than served.
@@ -124,6 +125,7 @@ impl WatchFlags {
     }
     pub fn readable(self) -> bool { self.0 & Self::READABLE.0 != 0 }
     pub fn writable(self) -> bool { self.0 & Self::WRITABLE.0 != 0 }
+    pub fn other_end_gone(self) -> bool { self.0 & Self::OTHER_END_GONE.0 != 0 }
     pub fn raw(self) -> u32 { self.0 }
 }
 
@@ -134,6 +136,7 @@ impl WatchFlags {
 struct Readiness {
     readable: bool,
     writable: bool,
+    other_end_gone: bool,
 }
 
 impl Readiness {
@@ -141,11 +144,12 @@ impl Readiness {
         let mut flags = 0u32;
         if self.readable { flags |= WatchFlags::READABLE.raw(); }
         if self.writable { flags |= WatchFlags::WRITABLE.raw(); }
+        if self.other_end_gone { flags |= WatchFlags::OTHER_END_GONE.raw(); }
         flags
     }
 
     fn any(self) -> bool {
-        self.readable || self.writable
+        self.readable || self.writable || self.other_end_gone
     }
 }
 
@@ -532,8 +536,17 @@ fn arm(
     let ready = readiness_of(object, flags).any();
     let read = if flags.readable() { ops::read_watch(object) } else { None };
     let write = if flags.writable() { ops::write_watch(object) } else { None };
-    // No readiness in either direction: nothing could ever answer this poll, so it is refused, not registered.
-    if !ready && read.is_none() && write.is_none() {
+    // A pipe end's own watch is the one its other end's last holder posts;
+    // an object with no other end is refused the question, whatever else the
+    // watch asks. A direction asked above has taken that watch already.
+    let own = match flags.other_end_gone().then(|| ops::pipe_end_watch(object)) {
+        Some(None) => return Err(SyscallError::NotSupported),
+        Some(Some(_)) if read.is_some() || write.is_some() => None,
+        Some(own) => own,
+        None => None,
+    };
+    // Nothing its interest names: nothing could ever answer this poll, so it is refused, not registered.
+    if !ready && read.is_none() && write.is_none() && own.is_none() {
         return Err(SyscallError::NotSupported);
     }
 
@@ -566,17 +579,21 @@ fn arm(
     if let Some(watch) = &write {
         watch.add_poll(PollEntry { poll: poll.clone(), direction: WatchFlags::WRITABLE });
     }
+    if let Some(watch) = &own {
+        watch.add_poll(PollEntry { poll: poll.clone(), direction: WatchFlags::OTHER_END_GONE });
+    }
     if readiness_of(object, flags).any() {
         poll.fire(0);
     }
     Ok(())
 }
 
-/// Per-direction readiness of the object, restricted to what was asked for.
+/// What the object answers each condition, restricted to what was asked for.
 fn readiness_of(object: &KObjectRef, flags: WatchFlags) -> Readiness {
     Readiness {
         readable: flags.readable() && ops::has_data(object),
         writable: flags.writable() && ops::has_space(object),
+        other_end_gone: flags.other_end_gone() && ops::other_end_gone(object),
     }
 }
 
diff --git a/kernel/src/object/ops.rs b/kernel/src/object/ops.rs
index 3ed962a6869..c1b1e1e2dc6 100644
--- a/kernel/src/object/ops.rs
+++ b/kernel/src/object/ops.rs
@@ -15,6 +15,7 @@ use crate::drivers::serial;
 use crate::file_cache;
 use crate::time::Deadline;
 use crate::pipe::{self, PipeId};
+use kernel::pipe::End;
 use crate::process::PipeMap;
 use crate::user_ptr::{UserBytes, UserBytesMut};
 use crate::inbox::PollEntry;
@@ -312,6 +313,32 @@ pub fn write_watch(object: &KObjectRef) -> Option<WatchRef> {
     }
 }
 
+/// The watch of a pipe end, which its other end's last holder posts as it
+/// lets go; `None` for an object that is no pipe end.
+pub fn pipe_end_watch(object: &KObjectRef) -> Option<WatchRef> {
+    match object {
+        KObjectRef::PipeRead(_) => read_watch(object),
+        KObjectRef::PipeWrite(_) => write_watch(object),
+        KObjectRef::Connection(_) | KObjectRef::Acceptor(_) | KObjectRef::Process(_)
+        | KObjectRef::Console(_) | KObjectRef::Device(_) | KObjectRef::SysCap(_)
+        | KObjectRef::File(_) | KObjectRef::Inbox(_) | KObjectRef::Connector(_)
+        | KObjectRef::Namespace(_) | KObjectRef::SharedMem(_) => None,
+    }
+}
+
+/// Whether this pipe end's other end has no holder left; `false` for an
+/// object that is no pipe end.
+pub fn other_end_gone(object: &KObjectRef) -> bool {
+    match object {
+        KObjectRef::PipeRead(r) => pipe::other_end_gone(r.id(), End::Read),
+        KObjectRef::PipeWrite(w) => pipe::other_end_gone(w.id(), End::Write),
+        KObjectRef::Connection(_) | KObjectRef::Acceptor(_) | KObjectRef::Process(_)
+        | KObjectRef::Console(_) | KObjectRef::Device(_) | KObjectRef::SysCap(_)
+        | KObjectRef::File(_) | KObjectRef::Inbox(_) | KObjectRef::Connector(_)
+        | KObjectRef::Namespace(_) | KObjectRef::SharedMem(_) => false,
+    }
+}
+
 /// Whether closing one handle to this object ends what its watches watch, so
 /// every poll on them — in any ring — is answered as gone. `false` for the log
 /// and the keyboard, which the machine ends on its own and which other handles
diff --git a/kernel/src/pipe.rs b/kernel/src/pipe.rs
index b0c8122a702..de41909970d 100644
--- a/kernel/src/pipe.rs
+++ b/kernel/src/pipe.rs
@@ -1,5 +1,6 @@
 use crate::mm::pmm;
 
+use kernel::pipe::{End, Holders, Released};
 use toyos_abi::ring::Ring;
 
 use alloc::sync::Arc;
@@ -39,7 +40,7 @@ pub use handle::{PipeReader, PipeWriter};
 /// take a counted reference, so no program point exists between the two where
 /// the last other end could close and free it.
 mod handle {
-    use super::{close_read, close_write, with_pipes_mut, Pipe, PipeId};
+    use super::{release, with_pipes_mut, End, Pipe, PipeId};
 
     /// One counted reference to a pipe's read end — `Arc`, for a reader slot.
     pub struct PipeReader(PipeId);
@@ -50,8 +51,7 @@ mod handle {
     impl PipeReader {
         /// The only constructor: taking and counting the reference is one statement.
         pub(super) fn acquire(pipe: &mut Pipe) -> Self {
-            pipe.readers = pipe.readers.checked_add(1).expect("pipe reader overflow");
-            pipe.publish_ends();
+            pipe.hold(End::Read);
             Self(pipe.id)
         }
 
@@ -60,8 +60,7 @@ mod handle {
 
     impl PipeWriter {
         pub(super) fn acquire(pipe: &mut Pipe) -> Self {
-            pipe.writers = pipe.writers.checked_add(1).expect("pipe writer overflow");
-            pipe.publish_ends();
+            pipe.hold(End::Write);
             Self(pipe.id)
         }
 
@@ -87,13 +86,13 @@ mod handle {
 
     impl Drop for PipeReader {
         fn drop(&mut self) {
-            close_read(self.0);
+            release(self.0, End::Read);
         }
     }
 
     impl Drop for PipeWriter {
         fn drop(&mut self) {
-            close_write(self.0);
+            release(self.0, End::Write);
         }
     }
 }
@@ -113,8 +112,7 @@ struct Pipe {
     id: PipeId,
     /// `None` until first use — allocating eagerly would charge every pending `SYS_CONNECT` 4 MiB before either end sent a byte.
     backing: Option<Backing>,
-    readers: u32,
-    writers: u32,
+    holders: Holders,
     /// The read end's and the write end's watches. Held by `Arc` so a blocking site or a
     /// poll registration can clone one out from under the table lock and hold it across its own park.
     read_watch: Arc<Watch>,
@@ -131,8 +129,7 @@ impl Pipe {
         Self {
             id,
             backing: None,
-            readers: 0,
-            writers: 0,
+            holders: Holders::new(),
             read_watch: Arc::new(Watch::new()),
             write_watch: Arc::new(Watch::new()),
             rt_boost_pending: false,
@@ -154,8 +151,13 @@ impl Pipe {
     /// Republish "is the other end gone?" into the mapped header, for netstack; the kernel itself decides from its own counts.
     fn publish_ends(&mut self) {
         let Some(backing) = self.backing.as_mut() else { return };
-        if self.readers == 0 { backing.ring.close_reader() } else { backing.ring.open_reader() }
-        if self.writers == 0 { backing.ring.close_writer() } else { backing.ring.open_writer() }
+        if self.holders.held(End::Read) { backing.ring.open_reader() } else { backing.ring.close_reader() }
+        if self.holders.held(End::Write) { backing.ring.open_writer() } else { backing.ring.close_writer() }
+    }
+
+    fn hold(&mut self, end: End) {
+        self.holders.hold(end);
+        self.publish_ends();
     }
 
     fn available(&self) -> u32 {
@@ -218,7 +220,7 @@ pub fn try_read(pipe_id: PipeId, buf: &mut UserBytesMut) -> Option<usize> {
             let boost = pipe.rt_boost_pending;
             pipe.rt_boost_pending = false;
             (Some(n), boost)
-        } else if pipe.writers == 0 {
+        } else if !pipe.holders.held(End::Write) {
             (Some(0), false)
         } else {
             (None, false)
@@ -241,7 +243,7 @@ pub enum PipeWrite {
 pub fn try_write(pipe_id: PipeId, buf: &UserBytes) -> Option<PipeWrite> {
     with_pipes_mut(|pipes| {
         let pipe = pipes.get_mut(pipe_id)?;
-        if pipe.readers == 0 {
+        if !pipe.holders.held(End::Read) {
             return Some(PipeWrite::BrokenPipe);
         }
         let Some(backing) = pipe.back() else {
@@ -257,16 +259,21 @@ pub fn try_write(pipe_id: PipeId, buf: &UserBytes) -> Option<PipeWrite> {
 
 pub fn has_data(pipe_id: PipeId) -> bool {
     with_pipes(|pipes| {
-        pipes.get(pipe_id).is_some_and(|p| p.available() > 0 || p.writers == 0)
+        pipes.get(pipe_id).is_some_and(|p| p.holders.readable(p.available()))
     })
 }
 
 pub fn has_space(pipe_id: PipeId) -> bool {
     with_pipes(|pipes| {
-        pipes.get(pipe_id).is_some_and(|p| p.space() > 0 || p.readers == 0)
+        pipes.get(pipe_id).is_some_and(|p| p.holders.writable(p.space()))
     })
 }
 
+/// Whether the end opposite `end` of this pipe has no holder left.
+pub fn other_end_gone(pipe_id: PipeId, end: End) -> bool {
+    with_pipes(|pipes| pipes.get(pipe_id).is_some_and(|p| p.holders.other_end_gone(end)))
+}
+
 /// Mark the pipe so the next consumer inherits RT priority.
 pub fn set_rt_boost_pending(pipe_id: PipeId) {
     with_pipes_mut(|pipes| {
@@ -276,41 +283,22 @@ pub fn set_rt_boost_pending(pipe_id: PipeId) {
     });
 }
 
-fn close_read(pipe_id: PipeId) {
-    // `true` when the pipe still lives and its write end is now the one to wake.
-    let wake_writers = with_pipes_mut(|pipes| {
-        let pipe = pipes.get_mut(pipe_id).expect("close_read: pipe not found");
-        pipe.readers = pipe.readers.checked_sub(1).expect("pipe reader underflow");
-        pipe.publish_ends();
-        if pipe.readers == 0 && pipe.writers == 0 {
-            let pipe = pipes.remove(pipe_id).unwrap();
-            free_pipe(pipe);
-            false // pipe freed, no one to wake
-        } else {
-            pipe.readers == 0
-        }
-    });
-    if wake_writers {
-        crate::scheduler::wake_pipe_writers(pipe_id);
-    }
-}
-
-fn close_write(pipe_id: PipeId) {
-    // `true` when the pipe still lives and its read end is now the one to wake.
-    let wake_readers = with_pipes_mut(|pipes| {
-        let pipe = pipes.get_mut(pipe_id).expect("close_write: pipe not found");
-        pipe.writers = pipe.writers.checked_sub(1).expect("pipe writer underflow");
+/// One reference to `end` lets go; its end's last posts the other end's watch.
+fn release(pipe_id: PipeId, end: End) {
+    let released = with_pipes_mut(|pipes| {
+        let pipe = pipes.get_mut(pipe_id).expect("release: pipe not found");
+        let released = pipe.holders.release(end);
         pipe.publish_ends();
-        if pipe.readers == 0 && pipe.writers == 0 {
+        if released == Released::PipeGone {
             let pipe = pipes.remove(pipe_id).unwrap();
             free_pipe(pipe);
-            false // pipe freed, no one to wake
-        } else {
-            pipe.writers == 0
         }
+        released
     });
-    if wake_readers {
-        crate::scheduler::wake_pipe_readers(pipe_id);
+    match (released, end) {
+        (Released::Held | Released::PipeGone, _) => {}
+        (Released::EndGone, End::Read) => crate::scheduler::wake_pipe_writers(pipe_id),
+        (Released::EndGone, End::Write) => crate::scheduler::wake_pipe_readers(pipe_id),
     }
 }
 
diff --git a/tests/netcase/system.toml b/tests/netcase/system.toml
index c8b1e20928e..4aaa4e89953 100644
--- a/tests/netcase/system.toml
+++ b/tests/netcase/system.toml
@@ -26,7 +26,10 @@ service = true
 serves = ["netstack"]
 devices = ["pci:1af4:1041"]
 
+# `netstack` because `netstack_socket_churn` connects through it and reads its
+# counts.
 [programs.test-runner]
+receives = ["netstack"]
 
 # The block service: the NVMe controller this machine's DATA is on, driven
 # from userland through its claim. Started again when it ends; the claim goes
diff --git a/tests/toyos-rust-tests/src/bin/netstack_socket_churn.rs b/tests/toyos-rust-tests/src/bin/netstack_socket_churn.rs
new file mode 100644
index 00000000000..f6d379b2834
--- /dev/null
+++ b/tests/toyos-rust-tests/src/bin/netstack_socket_churn.rs
@@ -0,0 +1,142 @@
+//! Piped TCP connections that end with no close request from their client, as
+//! a client that died leaves them: once netstack has let each connection go,
+//! its stream count is what it was before the first, and so is its count of
+//! sockets no table entry names.
+//!
+//! argv[1] is the port of the harness's host server, which ends every
+//! connection it accepts at once. Each round reads the peer's end of stream,
+//! drops both pipe ends, and waits for netstack's own count of live piped
+//! connections to return.
+//!
+//! argv[2] is the port of a second, which holds every connection it accepts
+//! and reads nothing. A client that shut its sending half down and left is one
+//! whose send pipe netstack never reads again, so nothing but the kernel's
+//! word on that pipe's other end can tell netstack it is gone: each such
+//! connection is counted ownerless, with the pipe empty and with bytes in it.
+//!
+//! On the same server, a client that moves netstack a handle that is no pipe
+//! end where its receive end belongs, and keeps its send end: the kernel
+//! refuses netstack's watch of it, and that refusal is all that ends the
+//! connection.
+
+use std::time::{Duration, Instant};
+
+use toyos::net::{MsgType, NetstackConn, TcpConnectPipedRequest, TcpConnectResponse};
+use toyos::OwnedHandle;
+use toyos_inspect::{Value, NET};
+
+/// The host, as QEMU's user network names it to a guest.
+const HOST: [u8; 4] = [10, 0, 2, 2];
+
+const ROUNDS: usize = 4;
+
+/// A hang ceiling on netstack letting one ended connection go, or counting
+/// one ownerless: it does the first on the pass after the peer acknowledges
+/// its FIN, and the second on the pass the client's leaving wakes.
+const LET_GO: Duration = Duration::from_secs(20);
+
+fn count(key: &str) -> u64 {
+    let snapshot = inspect::ask(NET).unwrap_or_else(|why| panic!("netstack's inspect answer: {why}"));
+    match snapshot.get(key) {
+        Some(Value::U64(n)) => *n,
+        other => panic!("netstack's snapshot carries {key} as {other:?}"),
+    }
+}
+
+/// A connection the peer holds open, whose client shuts its sending half down,
+/// leaves `unread` in a send pipe netstack no longer reads, and is gone.
+fn left_after_a_shutdown(port: u16, unread: &[u8]) {
+    let ownerless = count("net.piped.ownerless");
+    let conn = toyos::net::tcp_connect(HOST, port, 0).unwrap_or_else(|e| panic!("the holding server: {e:?}"));
+    toyos::net::tcp_shutdown(conn.socket_id, 1).unwrap_or_else(|e| panic!("shutting the sending half: {e:?}"));
+    if !unread.is_empty() {
+        assert_eq!(conn.tx.write_nonblock(unread), Ok(unread.len()), "bytes into a send pipe nobody reads");
+    }
+    drop(conn);
+    let asked = Instant::now();
+    while count("net.piped.ownerless") != ownerless + 1 {
+        assert!(
+            asked.elapsed() < LET_GO,
+            "netstack was not told a client that shut its sending half down, left {} byte(s) in its send pipe \
+             and dropped both ends is gone",
+            unread.len()
+        );
+    }
+}
+
+/// A connection the peer holds open, whose client keeps its send end and moved
+/// netstack an acceptor for the end netstack writes the peer's bytes into. The
+/// peer sends none, so netstack never writes it: its watch is refused, and the
+/// connection is reset and let go.
+fn a_receive_end_that_is_no_pipe_end_is_refused(port: u16) {
+    let live = count("net.piped.live");
+    let (acceptor, _connector) = toyos::port::create().expect("a port");
+    // SAFETY: the acceptor's one handle, which nothing else answers for.
+    let no_pipe_end = unsafe { OwnedHandle::from_raw(acceptor.into_raw()) };
+    let (from_client, kept) = toyos::pipe_pair().expect("a pipe");
+    let connected: TcpConnectResponse = NetstackConn::connect()
+        .and_then(|netstack| {
+            netstack.request_with_handles(
+                [no_pipe_end, OwnedHandle::from(from_client)],
+                MsgType::TcpConnectPiped,
+                &TcpConnectPipedRequest { addr: HOST, port, _pad: 0, timeout_ms: 0 },
+            )
+        })
+        .and_then(|pending| pending.response())
+        .unwrap_or_else(|e| panic!("the holding server, with an acceptor for a receive end: {e:?}"));
+    let id = connected.socket_id;
+    let asked = Instant::now();
+    while count("net.piped.live") != live {
+        assert!(
+            asked.elapsed() < LET_GO,
+            "netstack still counts connection {id} live {LET_GO:?} after the kernel refused the watch of its \
+             receive end, which is no pipe end"
+        );
+    }
+    drop(kept);
+}
+
+fn main() {
+    let port = |at: usize| -> u16 {
+        std::env::args()
+            .nth(at)
+            .and_then(|p| p.parse().ok())
+            .expect("usage: netstack_socket_churn <ending host port> <holding host port>")
+    };
+    let (port, holding) = (port(1), port(2));
+    let (streams, live) = (count("net.sockets.tcp"), count("net.piped.live"));
+    let untabled = count("net.sockets.untabled");
+    for round in 1..=ROUNDS {
+        let conn = toyos::net::tcp_connect(HOST, port, 0)
+            .unwrap_or_else(|e| panic!("connection {round} to the host server: {e:?}"));
+        let mut byte = [0u8; 1];
+        assert_eq!(conn.rx.read(&mut byte), Ok(0), "connection {round}: the host ends it without a byte");
+        // Both pipe ends, and no close request.
+        drop(conn);
+        let asked = Instant::now();
+        // Each question is a round trip netstack answers on a pass of its own.
+        while count("net.piped.live") != live {
+            assert!(
+                asked.elapsed() < LET_GO,
+                "netstack still counts connection {round} live {LET_GO:?} after both its ends closed"
+            );
+        }
+    }
+    let left = count("net.sockets.tcp");
+    println!(
+        "netstack_socket_churn: {ROUNDS} connections ended and let go; netstack held {streams} stream(s) before \
+         them and holds {left} after"
+    );
+    assert_eq!(left, streams, "netstack keeps a stream for a connection it let go");
+    assert_eq!(
+        count("net.sockets.untabled"),
+        untabled,
+        "netstack keeps the socket of a connection whose table entry it let go"
+    );
+    a_receive_end_that_is_no_pipe_end_is_refused(holding);
+    println!("netstack_socket_churn: a connection whose receive end is no pipe end was reset");
+    left_after_a_shutdown(holding, &[]);
+    left_after_a_shutdown(holding, b"never read");
+    println!("netstack_socket_churn: two clients that shut down and left are ownerless");
+    println!("netstack_socket_churn: ok");
+}
diff --git a/tests/toyos-rust-tests/src/bin/pipe_other_end_gone.rs b/tests/toyos-rust-tests/src/bin/pipe_other_end_gone.rs
new file mode 100644
index 00000000000..f67c2e683ec
--- /dev/null
+++ b/tests/toyos-rust-tests/src/bin/pipe_other_end_gone.rs
@@ -0,0 +1,198 @@
+//! A pipe end's `OTHER_END_GONE` watch: when it wakes, what it refuses, and
+//! what it does not take for a leaving.
+//!
+//! **The wake.** A thread parks on a ring that holds one `OTHER_END_GONE`
+//! watch and nothing else, and the other end's last holder goes: a reader is
+//! told of its writer and a writer of its reader, with the ring empty and with
+//! bytes left in it, and when the last holder was a process that ended. The
+//! leaving waits for the park, as `inbox_cancel_wakes`'s close does, and a
+//! waiter left parked is a hang the runner's ceiling reds. A watch made after
+//! the leaving is answered by the submit that makes it.
+//!
+//! **The refusal.** A connection and a file have no other end, and a watch that
+//! asks for one is refused whatever else it asks.
+//!
+//! **Not gone.** The other end has a holder while a duplicate of its handle is
+//! held, while the handle sits unreceived in a connection's queue, and after
+//! it moved to a child that still lives: the watch stays armed through each,
+//! and is answered once that holder goes.
+
+use std::io::Read;
+use std::os::toyos::process::CommandExt;
+use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
+
+use toyos::endow::{Endowments, SYSCAP_LABEL};
+use toyos::ipc::Connection;
+use toyos::poller::{Poller, OTHER_END_GONE, READABLE};
+use toyos::syscap::SysCap;
+use toyos::{namespace, pipe_pair, port, AsHandle, OwnedHandle, Pipe};
+use toyos_abi::syscall::{self, OpenFlags, SyscallError};
+use toyos_abi::RawHandle;
+
+#[path = "../roster.rs"]
+mod roster;
+
+const SELF_PATH: &str = "/system/bin/test_rs_pipe_other_end_gone";
+const TOKEN: u64 = 7;
+/// What the holding child is endowed the write end as. It never takes it: its
+/// table holds it.
+const HELD_LABEL: &str = "held";
+const SERVICE: &str = "queue";
+
+fn main() {
+    match std::env::args().nth(1).as_deref() {
+        Some("hold") => hold(),
+        Some(other) => panic!("unknown role {other:?}"),
+        None => test(),
+    }
+}
+
+/// Holds what it was endowed until its standard input ends.
+fn hold() {
+    let mut rest = Vec::new();
+    std::io::stdin().read_to_end(&mut rest).expect("the parent's end of standard input");
+}
+
+fn test() {
+    let cap: SysCap = Endowments::get()
+        .take(SYSCAP_LABEL)
+        .expect("test-runner endows every binary it spawns a system capability");
+
+    for unread in ["", "left in the ring"] {
+        let (reader, writer) = pipe(unread);
+        woken(&cap, &reader, || drop(writer));
+        let (reader, writer) = pipe(unread);
+        woken(&cap, &writer, || drop(reader));
+    }
+    println!("  a parked reader was told of its writer and a parked writer of its reader");
+
+    let (client, server) = connection();
+    refused(client.as_handle(), "a connection");
+    let file = syscall::open(SELF_PATH.as_bytes(), OpenFlags::READ).expect("this binary's own file");
+    refused(file, "a file");
+    syscall::close(file);
+    println!("  a connection and a file were refused the question");
+
+    let (reader, writer) = pipe("");
+    let second = syscall::dup(writer.as_handle()).expect("a second handle to the write end");
+    drop(writer);
+    held(&reader, "a duplicate of its handle is held");
+    syscall::close(second);
+    assert_eq!(at_last(&reader), Ok(OTHER_END_GONE), "the write end's last handle closed");
+
+    let (reader, writer) = pipe("");
+    client.send_handles([OwnedHandle::from(writer)]).expect("the write end into the connection's queue");
+    held(&reader, "its handle sits unreceived in a connection's queue");
+    let [received] = server.recv_handles_exact::<1>().expect("the handle the queue held");
+    held(&reader, "its handle was received");
+    syscall::close(received);
+    assert_eq!(at_last(&reader), Ok(OTHER_END_GONE), "the received handle closed");
+
+    let (reader, writer) = pipe("");
+    let mut child = Command::new(SELF_PATH)
+        .arg("hold")
+        .endow(HELD_LABEL, writer.into_raw().0)
+        .stdin(Stdio::piped())
+        .spawn()
+        .expect("spawn the holder");
+    let stdin = child.stdin.take().expect("the holder's standard input");
+    held(&reader, "its handle moved to a child that still lives");
+    woken(&cap, &reader, || drop(stdin));
+    assert!(child.wait().expect("wait the holder").success(), "the holder exited nonzero");
+
+    println!("pipe_other_end_gone: an end is told when its other end's last holder goes, and not before");
+}
+
+/// A fresh pipe with `unread` written into it.
+fn pipe(unread: &str) -> (Pipe, Pipe) {
+    let (reader, writer) = pipe_pair().expect("a pipe");
+    if !unread.is_empty() {
+        assert_eq!(writer.write_nonblock(unread.as_bytes()), Ok(unread.len()), "bytes into an empty pipe");
+    }
+    (reader, writer)
+}
+
+/// Both ends of one connection, made through a port of this process's own.
+fn connection() -> (Connection, Connection) {
+    let (acceptor, connector) = port::create().expect("a port");
+    let names = namespace::build().add(SERVICE, &connector).finish().expect("a namespace");
+    let client = names.open(SERVICE).expect("a connection to the port");
+    let server = acceptor.accept().expect("the connection's other end");
+    (client, server)
+}
+
+/// What one watch of `flags` on `handle` is answered by a submit that waits
+/// until `min_complete` answers are in: `None` while the kernel keeps it armed.
+fn watched(handle: RawHandle, flags: u32, min_complete: u32, timeout: u64) -> Option<Result<u32, SyscallError>> {
+    let poller = Poller::new(1);
+    poller.watch_raw(handle, flags, TOKEN);
+    let mut answer: Option<Result<u32, SyscallError>> = None;
+    poller.wait_answers(min_complete, timeout, |token, said| {
+        assert_eq!(token, TOKEN, "an answer under a token nobody gave");
+        assert!(answer.replace(said).is_none(), "one watch was answered twice");
+    });
+    answer
+}
+
+/// A watch for `end`'s other end leaving stays armed: the end has a holder.
+fn held(end: &Pipe, because: &str) {
+    assert_eq!(
+        watched(end.as_handle(), OTHER_END_GONE, 0, 0),
+        None,
+        "an end was told its other end is gone while {because}"
+    );
+}
+
+/// What a watch for `end`'s other end leaving is answered, however long that
+/// takes: a release is finished by whichever CPU drains it.
+fn at_last(end: &Pipe) -> Result<u32, SyscallError> {
+    watched(end.as_handle(), OTHER_END_GONE, 1, u64::MAX).expect("a wait with no deadline returned with no answer")
+}
+
+/// `handle` names no pipe end, so a watch that asks for its other end is
+/// refused, alone and beside a condition the handle does answer.
+fn refused(handle: RawHandle, what: &str) {
+    for flags in [OTHER_END_GONE, READABLE | OTHER_END_GONE] {
+        assert_eq!(
+            watched(handle, flags, 0, 0),
+            Some(Err(SyscallError::NotSupported)),
+            "a watch of {flags:#x} on {what}, which has no other end"
+        );
+    }
+}
+
+/// A thread parked on one `OTHER_END_GONE` watch of `end`, and on nothing
+/// else, is woken by `leave`, which lets the other end's last holder go; and a
+/// watch made after that is answered at once.
+fn woken(cap: &SysCap, end: &Pipe, leave: impl FnOnce()) {
+    let registered = AtomicBool::new(false);
+    let answers = thread::scope(|s| {
+        let waiter = s.spawn(|| {
+            let poller = Poller::new(1);
+            poller.watch(end, OTHER_END_GONE, TOKEN);
+            let mut answers = Vec::new();
+            // A non-blocking enter, so the watch is the kernel's before the
+            // other end goes.
+            poller.wait_answers(0, 0, |token, said| answers.push((token, said)));
+            assert!(answers.is_empty(), "an end was told its other end is gone while it is held: {answers:?}");
+            registered.store(true, Ordering::Release);
+            poller.wait_answers(1, u64::MAX, |token, said| answers.push((token, said)));
+            answers
+        });
+        // The roster first: its syscall is the loop's preemption point.
+        roster::await_true(|| {
+            roster::my_threads(cap).iter().any(|&(is_thread, state)| is_thread && state == roster::BLOCKED)
+                && registered.load(Ordering::Acquire)
+        });
+        leave();
+        waiter.join().expect("the waiter thread panicked")
+    });
+    assert_eq!(answers, [(TOKEN, Ok::<_, SyscallError>(OTHER_END_GONE))], "what the parked waiter was woken with");
+    assert_eq!(
+        watched(end.as_handle(), OTHER_END_GONE, 0, 0),
+        Some(Ok(OTHER_END_GONE)),
+        "a watch made after the other end's last holder left"
+    );
+}
diff --git a/tests/toyos.rs b/tests/toyos.rs
index b0ebf22de57..f5a648f743c 100644
--- a/tests/toyos.rs
+++ b/tests/toyos.rs
@@ -132,6 +132,9 @@ const RUST_SKIP: &[&str] = &[
     // launches ask DATA's server while this job's share holds all it may: the
     // `fs_share` metal row runs it on tests/proctreecase.
     "fs_share",
+    // It needs a NIC in front of netstack and a host server behind it:
+    // `netstack_socket_churn` runs it on `tests/netcase`.
+    "netstack_socket_churn",
     // It asserts nothing at all: it holds a `tests/lanleasecase` boot open for
     // twenty seconds. On a shared boot it would be twenty seconds of nothing.
     "lan_hold",
@@ -254,6 +257,10 @@ const MACHINE_TESTS: &[&str] = &[
     // behind its emulated VT-d unit and without one: no shipped machine has a
     // virtio function, so only a QEMU machine can be asked.
     "iommu_virtio_platform",
+    // netstack's own state behind a real stack and a peer that ends its
+    // connections: netstack is one binary that owns its NIC, with no host
+    // build, and the T14's peer is the bench's network.
+    "netstack_socket_churn",
     // The nested-NMI report is a raw write to the 16550, which the T14 does not
     // have.
     "nested_nmi_is_loud",
@@ -2786,11 +2793,50 @@ fn scanout_wc(console: &str) -> Result<(), String> {
     Ok(())
 }
 
+/// netstack's stream count returns once connections that ended without their
+/// client's close request are let go, a client that left a connection its
+/// peer holds is counted gone, and a connection whose receive end the kernel
+/// refuses netstack's watch of is reset. One host server here ends each
+/// connection it accepts at once and one holds each; the guest's comparisons
+/// are the verdict.
+fn netstack_socket_churn() -> Result<(), String> {
+    const JOB: &str = "netstack_socket_churn";
+    const LEASED: &str = "netstack: DHCP: lease ";
+    let server = std::net::TcpListener::bind(("127.0.0.1", 0)).map_err(|e| format!("the host server: {e}"))?;
+    let port = server.local_addr().map_err(|e| format!("the host server's port: {e}"))?.port();
+    // Ends with the process: a guest that never dials leaves it in `accept`.
+    thread::spawn(move || server.incoming().for_each(drop));
+    // A second that holds what it accepts and reads none of it, for as long
+    // as the process lives.
+    let holder = std::net::TcpListener::bind(("127.0.0.1", 0)).map_err(|e| format!("the holding server: {e}"))?;
+    let holding = holder.local_addr().map_err(|e| format!("the holding server's port: {e}"))?.port();
+    thread::spawn(move || holder.incoming().collect::<Vec<_>>());
+
+    let bin = qemu::build_toyos_bin(qemu::SUITE_ARCH, &compile::repo_root().join("tests/toyos-rust-tests"), JOB);
+    let case = compile::repo_root().join("tests/netcase");
+    let mut qemu = QemuInstance::boot_with_options(&case, &[], &[(JOB.to_string(), bin)], BootOptions::default());
+    let mut console = qemu.boot_log().to_string();
+    await_marker(&mut qemu, &mut console, LEASED, "netstack's lease").map_err(|e| format!("{e}\n{console}"))?;
+    let result =
+        qemu.run_test(&format!("test_rs_netstack_socket_churn {port} {holding}"), Duration::from_secs(120));
+    if let Some(why) = &result.error {
+        return Err(format!("{why}\nthe job said:\n{}", result.stdout));
+    }
+    if result.exit_code != Some(0) {
+        return Err(format!("the job ended {:?}:\n{}", result.exit_code, result.stdout));
+    }
+    if !result.stdout.lines().any(|l| l.trim_end().ends_with("netstack_socket_churn: ok")) {
+        return Err(format!("the guest never said it was done:\n{}", result.stdout));
+    }
+    Ok(())
+}
+
 /// Run the machine-shape test, which owns its QEMU: the machine shape *is* the
 /// test.
 fn run_machine_test(name: &str, test_config: &Path) -> Result<(), String> {
     match name {
         "iommu_virtio_platform" => common::iommu::iommu_virtio_platform(test_config),
+        "netstack_socket_churn" => netstack_socket_churn(),
         "nested_nmi_is_loud" => faults::nested_nmi_is_loud(test_config),
         "machine_shutdown" => power::machine_shutdown(test_config),
         "acpi_power_button" => power::acpi_power_button(test_config),
diff --git a/toyos-abi/src/inbox.rs b/toyos-abi/src/inbox.rs
index a8955708fee..c1a1754522c 100644
--- a/toyos-abi/src/inbox.rs
+++ b/toyos-abi/src/inbox.rs
@@ -24,10 +24,14 @@ pub const OP_ACCEPT: u8 = 3;
 
 /// Readiness flags for [`OP_WATCH`], stored in `Submission::op_flags`.
 ///
-/// Honest at both ends: the same two bits are the interest going in and the
+/// Honest at both ends: the same bits are the interest going in and the
 /// result coming back in `Completion::result`.
 pub const READABLE: u32 = 1;
 pub const WRITABLE: u32 = 4;
+/// Of a pipe end: the pipe's other end has no holder left, in any process,
+/// whatever the pipe still holds. It stays so. A handle that is no pipe end
+/// has no other end, and a watch that asks this of it is refused.
+pub const OTHER_END_GONE: u32 = 16;
 
 /// One piece of work. Written by userspace into the submission array.
 #[repr(C)]
diff --git a/toyos/src/poller.rs b/toyos/src/poller.rs
index f7df1ec4e35..5a724a24fc0 100644
--- a/toyos/src/poller.rs
+++ b/toyos/src/poller.rs
@@ -2,14 +2,14 @@
 
 use core::sync::atomic::{AtomicU32, Ordering};
 use toyos_abi::RawHandle;
-use toyos_abi::syscall;
+use toyos_abi::syscall::{self, SyscallError};
 use toyos_abi::inbox::{
     Submission, Completion, RingHeader, RingLayout,
     OP_WATCH, SUBMISSION_RING_OFF, COMPLETION_RING_OFF, SUBMISSIONS_OFF,
 };
 use crate::AsHandle;
 
-pub use toyos_abi::inbox::{READABLE, WRITABLE};
+pub use toyos_abi::inbox::{OTHER_END_GONE, READABLE, WRITABLE};
 
 /// The inbox page, and the only thing in this crate that touches it.
 ///
@@ -248,7 +248,7 @@ impl Poller {
 
     /// Watch the given handle for readiness.
     ///
-    /// `flags` are [`READABLE`] / [`WRITABLE`].
+    /// `flags` are [`READABLE`] / [`WRITABLE`] / [`OTHER_END_GONE`].
     /// `token` is returned in completions to identify which handle is ready.
     pub fn watch(&self, handle: &impl AsHandle, flags: u32, token: u64) {
         self.watch_raw(handle.as_handle(), flags, token);
@@ -307,6 +307,18 @@ impl Poller {
     /// Blocks until at least `min_complete` completions are ready or `timeout_nanos`
     /// elapses. Calls `f` for each completed token.
     pub fn wait(&self, min_complete: u32, timeout_nanos: u64, mut f: impl FnMut(u64)) {
+        self.wait_answers(min_complete, timeout_nanos, |token, _| f(token));
+    }
+
+    /// [`wait`](Self::wait), with what each watch was answered beside its
+    /// token: the conditions its handle met, or the kernel's word for why the
+    /// watch is over and will not fire.
+    pub fn wait_answers(
+        &self,
+        min_complete: u32,
+        timeout_nanos: u64,
+        mut f: impl FnMut(u64, Result<u32, SyscallError>),
+    ) {
         self.submit(min_complete, timeout_nanos);
         self.drain(&mut f);
     }
@@ -315,7 +327,7 @@ impl Poller {
     ///
     /// Split from [`wait`](Self::wait) because it is the half that is a pure
     /// function of the page: a host test can hand it a fake one.
-    fn drain(&self, f: &mut impl FnMut(u64)) {
+    fn drain(&self, f: &mut impl FnMut(u64, Result<u32, SyscallError>)) {
         // Unreachable, and kept for that reason: `capacity` bounds the
         // registrations and the rings are sized from `capacity`, so nothing a
         // conforming caller does can make the kernel drop a completion here.
@@ -345,7 +357,13 @@ impl Poller {
             // to that exactly as to readiness — by looking at the handle again.
             // A zero result is meaningful too: `OP_ACCEPT` reports handle 0
             // that way.
-            f(completion.token);
+            // The kernel writes a refusal as its error's number, negated.
+            let refusal = u64::MAX - u64::from(completion.result.unsigned_abs());
+            let answer = match completion.result {
+                met @ 1.. => Ok(met as u32),
+                _ => Err(SyscallError::from_u64(refusal).unwrap_or(SyscallError::Unknown)),
+            };
+            f(completion.token, answer);
             self.rings.completion_head().store(head.wrapping_add(1), Ordering::Release);
         }
     }
diff --git a/userland/netstack/src/listen/tests.rs b/userland/netstack/src/listen/tests.rs
index 856a9816e0d..da0179aac32 100644
--- a/userland/netstack/src/listen/tests.rs
+++ b/userland/netstack/src/listen/tests.rs
@@ -3,7 +3,9 @@
 //! is what the port answers the next peer.
 
 use super::*;
+use crate::{Ownerless, OWNERLESS_LIFE, RESET_LIFE};
 use std::collections::VecDeque;
+use std::time::Duration;
 
 use smoltcp::iface::{Config, Interface, PollResult, SocketHandle, SocketSet};
 use smoltcp::phy::{self, ChecksumCapabilities, Device, DeviceCapabilities, Medium};
@@ -84,6 +86,10 @@ struct Net {
     listener: SocketHandle,
     listening: Listening,
     sent: Vec<Sent>,
+    /// The clock every poll reads.
+    now: Instant,
+    /// Whether the far end answers ARP.
+    arp: bool,
 }
 
 impl Net {
@@ -95,7 +101,16 @@ impl Net {
         socket.listen(PORT).expect("a fresh socket listens");
         let mut sockets = SocketSet::new(Vec::new());
         let listener = sockets.add(socket);
-        Self { iface, wire, sockets, listener, listening: Listening::new(PORT), sent: Vec::new() }
+        Self {
+            iface,
+            wire,
+            sockets,
+            listener,
+            listening: Listening::new(PORT),
+            sent: Vec::new(),
+            now: Instant::from_millis(0),
+            arp: true,
+        }
     }
 
     fn socket(&mut self) -> &mut tcp::Socket<'static> {
@@ -103,10 +118,10 @@ impl Net {
     }
 
     /// One pass of netstack's loop: everything the wire holds, in one batch, and
-    /// the far end's ARP answered.
+    /// the far end's ARP answered while it answers any.
     fn pass(&mut self) {
         loop {
-            while self.iface.poll(Instant::from_millis(0), &mut self.wire, &mut self.sockets) != PollResult::None {}
+            while self.iface.poll(self.now, &mut self.wire, &mut self.sockets) != PollResult::None {}
             if self.wire.outbound.is_empty() {
                 return;
             }
@@ -122,6 +137,9 @@ impl Net {
             EthernetProtocol::Arp => {
                 let arp = ArpRepr::parse(&ArpPacket::new_checked(eth.payload()).unwrap()).unwrap();
                 let ArpRepr::EthernetIpv4 { operation: ArpOperation::Request, .. } = arp else { return };
+                if !self.arp {
+                    return;
+                }
                 let reply = ArpRepr::EthernetIpv4 {
                     operation: ArpOperation::Reply,
                     source_hardware_addr: PEER_MAC,
@@ -312,3 +330,194 @@ fn an_accept_refused_for_room_is_woken_again_when_room_returns() {
     assert!(net.wakes(true), "an owner refused for room was never woken again");
     assert_eq!(net.accept(true, true), Accept::Take(()));
 }
+
+/// **A connection netstack aborts keeps its socket until the reset has left.**
+/// A socket taken out of the set on the pass that aborted it sends nothing,
+/// and its peer holds a connection nobody will ever answer.
+#[test]
+fn an_aborted_connection_is_spent_once_its_reset_has_left() {
+    let mut net = Net::new();
+    let isn = net.syn(5001);
+    net.send(5001, TcpControl::None, PEER_ISN + 1, Some(isn + 1));
+    net.pass();
+    assert!(!crate::spent(net.socket()), "an established connection was spent");
+    net.socket().abort();
+    assert!(!crate::spent(net.socket()), "an aborted connection was spent with its reset still owed");
+    net.pass();
+    let last = net.sent.last().expect("an abort is said on the wire");
+    assert_eq!((last.control, last.to), (TcpControl::Rst, 5001));
+    assert!(crate::spent(net.socket()), "a connection whose reset has left was kept");
+}
+
+/// A connection both ends have closed is spent while it only waits out
+/// `TimeWait`, and not while its own FIN is unanswered.
+#[test]
+fn a_connection_closed_by_both_ends_is_spent() {
+    let mut net = Net::new();
+    let isn = net.syn(5001);
+    net.send(5001, TcpControl::None, PEER_ISN + 1, Some(isn + 1));
+    net.pass();
+    net.socket().close();
+    net.pass();
+    assert!(!crate::spent(net.socket()), "a connection was spent with its FIN unanswered");
+    net.send(5001, TcpControl::Fin, PEER_ISN + 1, Some(isn + 2));
+    net.pass();
+    assert_eq!(net.socket().state(), tcp::State::TimeWait, "the premise: the peer's FIN answered ours");
+    assert!(crate::spent(net.socket()), "a connection both ends closed was kept");
+    assert_eq!(crate::ownerless(net.socket(), Duration::ZERO, false), Ownerless::Over);
+}
+
+/// The handshake from `from` finished: a connection a client would hold.
+fn established(net: &mut Net, from: u16) -> u32 {
+    let isn = net.syn(from);
+    net.send(from, TcpControl::None, PEER_ISN + 1, Some(isn + 1));
+    net.pass();
+    assert_eq!(net.socket().state(), tcp::State::Established);
+    isn
+}
+
+/// A second at a time from the client's leaving to [`OWNERLESS_LIFE`], the peer
+/// doing `each_second`: the connection waits for all of it, and what it is at
+/// the end is the answer.
+fn at_the_ceiling(net: &mut Net, mut each_second: impl FnMut(&mut Net)) -> Ownerless {
+    for second in 0..OWNERLESS_LIFE.as_secs() {
+        each_second(net);
+        net.pass();
+        let gone = Duration::from_secs(second);
+        assert_eq!(
+            crate::ownerless(net.socket(), gone, false),
+            Ownerless::Waits,
+            "after {second}s in {}",
+            net.socket().state()
+        );
+        net.now += smoltcp::time::Duration::from_secs(1);
+    }
+    net.pass();
+    crate::ownerless(net.socket(), OWNERLESS_LIFE, false)
+}
+
+/// The reset [`Ownerless::Cut`] owes leaves at the next poll, and the
+/// connection is over.
+fn the_cut_is_said(net: &mut Net) {
+    net.pass();
+    let last = net.sent.last().expect("a cut connection's reset");
+    assert_eq!((last.control, last.to), (TcpControl::Rst, 5001));
+    assert_eq!(crate::ownerless(net.socket(), Duration::ZERO, true), Ownerless::Over);
+}
+
+/// The half of netstack's pass that comes before its bridge: the interface
+/// polled until it has nothing to do, and nothing read off the wire since.
+/// What it sent stays on the wire for [`Net::pass`] to answer.
+fn poll_only(net: &mut Net) {
+    while net.iface.poll(net.now, &mut net.wire, &mut net.sockets) != PollResult::None {}
+}
+
+/// A connection in `FIN-WAIT-2` whose peer acknowledged the FIN once and then
+/// said nothing, cut at the ceiling: its next hop's neighbour entry ran out
+/// forty seconds before.
+fn cut_after_a_silence(net: &mut Net) {
+    let isn = established(net, 5001);
+    net.socket().close();
+    net.pass();
+    net.send(5001, TcpControl::None, PEER_ISN + 1, Some(isn + 2));
+    assert_eq!(at_the_ceiling(net, |_| {}), Ownerless::Cut);
+}
+
+/// **A cut connection is kept until its reset has left.** The pass after the
+/// cut asks the next hop for its address and sends nothing else; the reset
+/// leaves on the pass that reads the answer.
+#[test]
+fn a_cut_connection_whose_neighbour_entry_ran_out_is_kept_until_its_reset_leaves() {
+    let mut net = Net::new();
+    cut_after_a_silence(&mut net);
+    let said = net.sent.len();
+    poll_only(&mut net);
+    let asked: Vec<_> =
+        net.wire.outbound.iter().map(|f| EthernetFrame::new_checked(&f[..]).unwrap().ethertype()).collect();
+    assert_eq!(asked, [EthernetProtocol::Arp], "the premise: the reset waits on the next hop's address");
+    assert_eq!(
+        crate::ownerless(net.socket(), Duration::ZERO, true),
+        Ownerless::Waits,
+        "a cut connection was let go before its next hop could answer"
+    );
+    net.pass();
+    assert_eq!(net.sent.len(), said + 1, "the next hop answered and the reset stayed");
+    the_cut_is_said(&mut net);
+}
+
+/// **And no longer than [`RESET_LIFE`] when the next hop answers nothing.**
+#[test]
+fn a_cut_connection_whose_next_hop_answers_no_arp_is_let_go_after_the_reset_bound() {
+    let mut net = Net::new();
+    cut_after_a_silence(&mut net);
+    net.arp = false;
+    let said = net.sent.len();
+    for second in 0..RESET_LIFE.as_secs() {
+        net.pass();
+        assert_eq!(
+            crate::ownerless(net.socket(), Duration::from_secs(second), true),
+            Ownerless::Waits,
+            "{second}s after the cut"
+        );
+        net.now += smoltcp::time::Duration::from_secs(1);
+    }
+    net.pass();
+    assert_eq!(crate::ownerless(net.socket(), RESET_LIFE, true), Ownerless::Unsaid, "the slot never comes back");
+    assert_eq!(net.sent.len(), said, "the premise: no segment leaves without a next hop");
+}
+
+/// **A connection netstack aborted before the ceiling is cut there too**, its
+/// reset still owed: the neighbour entry the handshake made has run out and
+/// the next hop answers no ARP.
+#[test]
+fn an_aborted_connection_whose_next_hop_answers_no_arp_is_cut_at_the_ceiling() {
+    let mut net = Net::new();
+    established(&mut net, 5001);
+    net.now += smoltcp::time::Duration::from_secs(61);
+    net.arp = false;
+    net.socket().abort();
+    let said = net.sent.len();
+    assert_eq!(at_the_ceiling(&mut net, |_| {}), Ownerless::Cut);
+    assert_eq!(crate::ownerless(net.socket(), RESET_LIFE, true), Ownerless::Unsaid);
+    assert_eq!(net.sent.len(), said, "the premise: no segment leaves without a next hop");
+}
+
+/// **A peer that acknowledges nothing of a full send buffer holds a connection
+/// with no client no longer than the ceiling.** The socket takes no more of
+/// the client's bytes for as long as it lives, so what its send pipe still
+/// holds is never read.
+#[test]
+fn a_peer_that_acknowledges_nothing_of_a_full_send_buffer_is_reset_at_the_ceiling() {
+    let mut net = Net::new();
+    established(&mut net, 5001);
+    assert_eq!(net.socket().send_slice(&[0u8; 4096]), Ok(4096), "the premise: the bench's send buffer is full");
+    let cut = at_the_ceiling(&mut net, |net| assert!(!crate::send_room(net.socket()), "the premise: no send room"));
+    assert_eq!(cut, Ownerless::Cut, "the slot of a connection with no client never comes back");
+    the_cut_is_said(&mut net);
+}
+
+/// **A peer that goes silent after the handshake holds a connection with no
+/// client no longer than the ceiling.** Its FIN is never acknowledged, and
+/// smoltcp retransmits it for as long as the socket lives.
+#[test]
+fn a_peer_gone_silent_is_reset_at_the_ceiling() {
+    let mut net = Net::new();
+    established(&mut net, 5001);
+    net.socket().close();
+    assert_eq!(at_the_ceiling(&mut net, |_| {}), Ownerless::Cut, "the slot of a connection with no client never comes back");
+    assert_eq!(net.socket().state(), tcp::State::Closed);
+    the_cut_is_said(&mut net);
+}
+
+/// **Nor does a peer that keeps answering.** It acknowledges the FIN, never
+/// sends its own, and says so again every second: `FinWait2` has no timer, and
+/// a socket timeout counted from the peer's last word would never run out.
+#[test]
+fn a_peer_that_answers_and_never_closes_is_reset_at_the_ceiling() {
+    let mut net = Net::new();
+    let isn = established(&mut net, 5001);
+    net.socket().close();
+    let answer = |net: &mut Net| net.send(5001, TcpControl::None, PEER_ISN + 1, Some(isn + 2));
+    assert_eq!(at_the_ceiling(&mut net, answer), Ownerless::Cut, "the slot of a connection with no client never comes back");
+    the_cut_is_said(&mut net);
+}
diff --git a/userland/netstack/src/main.rs b/userland/netstack/src/main.rs
index 0e611b67c40..fd74c9de52b 100644
--- a/userland/netstack/src/main.rs
+++ b/userland/netstack/src/main.rs
@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 use std::time::{Duration, Instant};
-use toyos::poller::{READABLE, WRITABLE, Poller};
+use toyos::poller::{OTHER_END_GONE, READABLE, WRITABLE, Poller};
 use toyos::ipc;
 use toyos::AsHandle;
 use toyos::ipc::{Connection, IpcPayload, RxStep};
@@ -331,10 +331,31 @@ struct PendingUdpRecv {
 }
 
 /// A piped TCP connection: data flows through kernel pipes instead of IPC messages.
+///
+/// **The socket and its id live exactly as long as this does.** What ends a
+/// connection is what the kernel says of the client's pipe ends and what the
+/// peer says on the wire; a close request only asks for that end early, and a
+/// client that dies sends none.
+///
+/// **A client is gone when the kernel says so of both its ends**
+/// ([`OTHER_END_GONE`], watched on each pipe for as long as netstack holds
+/// it), whatever its send pipe still holds and whether or not the socket
+/// takes bytes. A direction netstack itself has closed counts as gone. Such a
+/// connection is [`ownerless`]: the wire has [`OWNERLESS_LIFE`] to finish it,
+/// and a reset that then cannot leave has [`RESET_LIFE`]. A client holding an
+/// end of a direction still open is never timed.
 struct PipedConnection {
+    socket_id: u32,
     handle: SocketHandle,
     rx_write: Option<Pipe>,
     tx_read: Option<Pipe>,
+    /// The kernel said no holder of the send pipe's write end is left. What
+    /// the pipe holds is still the peer's.
+    writer_gone: bool,
+    /// When a pass first found the client gone, or cut the connection.
+    ownerless: Option<Instant>,
+    /// [`Ownerless::Cut`] was this connection's answer.
+    cut: bool,
     /// The client's receive pipe refused bytes the socket still holds, so the
     /// pipe is watched for room.
     held: bool,
@@ -344,9 +365,9 @@ impl PipedConnection {
     /// **A client's handle that refuses netstack for any reason but a full pipe or
     /// a vanished reader ends that client's connection, never netstack.** The
     /// ends are whatever the client moved, and nothing checks their kind at
-    /// intake: a read end, a file past its size limit or a handle with no
-    /// `WRITE` right each answer a refusal here. So does a pipe whose ring
-    /// page could not be allocated, which no wait cures.
+    /// intake: a read end or a handle with no `WRITE` right answers a refusal
+    /// here, and one that is no pipe end has its watch refused. So does a pipe
+    /// whose ring page could not be allocated, which no wait cures.
     fn refuse(&mut self, socket: &mut tcp::Socket, end: &str, e: toyos_abi::syscall::SyscallError) {
         say!("netstack: resetting a connection — its {end} pipe refused netstack: {e:?}");
         socket.abort();
@@ -366,8 +387,29 @@ impl PipedConnection {
         self.close_tx();
     }
 
-    fn is_fully_closed(&self) -> bool {
-        self.rx_write.is_none() && self.tx_read.is_none()
+    /// The client holds no end of a direction that is still open.
+    fn clientless(&self) -> bool {
+        self.rx_write.is_none() && (self.tx_read.is_none() || self.writer_gone)
+    }
+
+    /// What the kernel answered a watch on the client's send pipe, or on its
+    /// receive pipe. **Of a pipe netstack still holds**: closing one ends its
+    /// watch, and that end is an answer too.
+    fn pipe_answered(
+        &mut self,
+        socket: &mut tcp::Socket,
+        send: bool,
+        answer: Result<u32, toyos_abi::syscall::SyscallError>,
+    ) {
+        let (held, end) = if send { (&self.tx_read, "send") } else { (&self.rx_write, "receive") };
+        match answer {
+            _ if held.is_none() => {}
+            Err(e) => self.refuse(socket, end, e),
+            Ok(met) if met & OTHER_END_GONE == 0 => {}
+            Ok(_) if send => self.writer_gone = true,
+            // Nobody is left to read it.
+            Ok(_) => self.close_rx(),
+        }
     }
 }
 
@@ -419,11 +461,71 @@ fn send_room(socket: &tcp::Socket) -> bool {
     socket.can_send() && socket.send_capacity() > socket.send_queue()
 }
 
-fn piped_connection(handle: SocketHandle, pipes: DataPipes) -> PipedConnection {
+/// Whether `socket` has said its last to its peer: it is closed, and the reset
+/// an abort owes has left. `TimeWait` only waits.
+fn spent(socket: &tcp::Socket) -> bool {
+    !socket.is_open() && !(socket.state() == tcp::State::Closed && socket.remote_endpoint().is_some())
+}
+
+/// How long the wire has to finish a connection whose client's pipe ends are
+/// both gone, before netstack resets it: R2, the time RFC 9293 §3.8.3 gives a
+/// segment's retransmission before the connection is closed, at the 100
+/// seconds it asks for at least.
+///
+/// **From the client's leaving and not from the peer's last word**, so a peer
+/// that keeps answering holds a slot no longer than one that says nothing:
+/// RFC 9293 §3.8.6.1 lets a system reclaim a connection its peer holds open.
+const OWNERLESS_LIFE: Duration = Duration::from_secs(100);
+
+/// How long the reset of a connection cut at [`OWNERLESS_LIFE`] has to leave.
+/// A connection that sent and heard nothing for that long has outlived its
+/// next hop's neighbour entry, so the reset waits on an ARP answer.
+///
+/// Address resolution's own budget: RFC 4861 §7.2.2 fails it after
+/// `MAX_MULTICAST_SOLICIT` solicitations `RETRANS_TIMER` apart, 3 and 1,000
+/// milliseconds in §10. That is IPv6's; RFC 1122 §2.3.2.1 gives ARP a rate of
+/// one request a second per destination and no count, and smoltcp asks at that
+/// rate for as long as the socket lives.
+const RESET_LIFE: Duration = Duration::from_secs(3);
+
+/// What a pass makes of a connection whose client is gone.
+#[derive(Debug, PartialEq, Eq)]
+enum Ownerless {
+    /// The wire still owes something, and has time left.
+    Waits,
+    /// Reset at [`OWNERLESS_LIFE`] with the wire unfinished, and kept until
+    /// the reset has left or [`RESET_LIFE`] is over.
+    Cut,
+    /// The wire is finished.
+    Over,
+    /// Let go with a reset that never left: no next hop took it.
+    Unsaid,
+}
+
+/// The pass's answer for `socket`, whose client has been gone for `waited`, or
+/// which was cut `waited` ago.
+fn ownerless(socket: &mut tcp::Socket, waited: Duration, cut: bool) -> Ownerless {
+    if spent(socket) {
+        Ownerless::Over
+    } else if waited < if cut { RESET_LIFE } else { OWNERLESS_LIFE } {
+        Ownerless::Waits
+    } else if cut {
+        Ownerless::Unsaid
+    } else {
+        socket.abort();
+        Ownerless::Cut
+    }
+}
+
+fn piped_connection(socket_id: u32, handle: SocketHandle, pipes: DataPipes) -> PipedConnection {
     PipedConnection {
+        socket_id,
         handle,
         rx_write: Some(pipes.to_client),
         tx_read: Some(pipes.from_client),
+        writer_gone: false,
+        ownerless: None,
+        cut: false,
         held: false,
     }
 }
@@ -531,8 +633,8 @@ const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(2);
 /// is counted down and discarded, never waited for.
 const MAX_KEPT_REQUEST: usize = 256;
 
-/// Registrations one piped connection can make in a batch: its tx pipe, and
-/// its rx pipe while that pipe is holding bytes back.
+/// Registrations one piped connection can make in a batch: its send pipe and
+/// its receive pipe.
 const POLL_HANDLES_PER_PIPED: u32 = 2;
 
 /// Registrations the lookups make in a batch: each waiting client's
@@ -729,6 +831,7 @@ impl Netstack {
         snap.put("sockets.listeners", listeners);
         snap.put("sockets.udp", udp);
         snap.put("piped.live", self.piped_live());
+        snap.put("piped.ownerless", self.piped_connections.iter().filter(|c| c.ownerless.is_some()).count());
         snap.put("piped.max", self.max_piped_connections);
     }
 
@@ -1246,7 +1349,7 @@ impl Netstack {
         let stream_id = self.alloc_id();
         self.sockets.insert(stream_id, SocketKind::TcpStream(old_handle));
 
-        self.piped_connections.push(piped_connection(old_handle, pipes));
+        self.piped_connections.push(piped_connection(stream_id, old_handle, pipes));
 
         // Create replacement listener
         let rx_buf = tcp::SocketBuffer::new(vec![0u8; TCP_SOCKET_BUFFER]);
@@ -1369,24 +1472,63 @@ impl Netstack {
                 conn.close_tx();
             }
 
-            // Detect client death: a zero-byte write is refused by name once
-            // the pipe has no reader — the kernel's fact, not the client's.
-            if let Some(ref pipe) = conn.rx_write {
-                match toyos_abi::syscall::write_nonblock(pipe.as_handle(), &[]) {
-                    Ok(_) | Err(SyscallError::WouldBlock) => {}
-                    Err(SyscallError::Gone) => conn.close_rx(),
-                    Err(e) => conn.refuse(socket, "receive", e),
+            if conn.clientless() {
+                let waited = conn.ownerless.get_or_insert_with(Instant::now).elapsed();
+                match ownerless(socket, waited, conn.cut) {
+                    Ownerless::Waits => {}
+                    Ownerless::Cut => {
+                        say!(
+                            "netstack: resetting a connection — its client left {}s ago and its peer has not finished it",
+                            waited.as_secs()
+                        );
+                        (conn.ownerless, conn.cut) = (Some(Instant::now()), true);
+                    }
+                    Ownerless::Over => {
+                        if conn.cut {
+                            say!("netstack: the reset has left");
+                        }
+                        closed.push(i);
+                    }
+                    Ownerless::Unsaid => {
+                        say!(
+                            "netstack: letting a connection go with its reset unsent — no next hop took it in {}s",
+                            RESET_LIFE.as_secs()
+                        );
+                        closed.push(i);
+                    }
                 }
             }
-
-            // Fully clean up when both sides are done
-            if conn.is_fully_closed() && !socket.is_open() {
-                closed.push(i);
-            }
         }
 
         for &i in closed.iter().rev() {
-            self.piped_connections.swap_remove(i);
+            let conn = self.piped_connections.swap_remove(i);
+            socket_set.remove(conn.handle);
+            self.sockets.remove(&conn.socket_id);
+        }
+    }
+
+    /// How long until the first connection with no client left reaches
+    /// [`OWNERLESS_LIFE`], or a cut one [`RESET_LIFE`], which nothing on the
+    /// wire wakes a pass for.
+    fn ownerless_wake_in(&self) -> Option<Duration> {
+        self.piped_connections
+            .iter()
+            .filter_map(|c| Some((c.ownerless?, if c.cut { RESET_LIFE } else { OWNERLESS_LIFE })))
+            .map(|(since, life)| life.saturating_sub(since.elapsed()))
+            .min()
+    }
+
+    /// What the kernel answered a watch on a pipe of connection `socket_id`,
+    /// which may be gone since: the watch of a pipe closed with it ends too.
+    fn pipe_answered(
+        &mut self,
+        socket_set: &mut SocketSet<'_>,
+        socket_id: u32,
+        send: bool,
+        answer: Result<u32, toyos_abi::syscall::SyscallError>,
+    ) {
+        if let Some(conn) = self.piped_connections.iter_mut().find(|c| c.socket_id == socket_id) {
+            conn.pipe_answered(socket_set.get_mut::<tcp::Socket>(conn.handle), send, answer);
         }
     }
 
@@ -1474,7 +1616,7 @@ impl Netstack {
                 };
                 pc.client.result(&resp);
                 let pc = self.pending_piped_connects.swap_remove(i);
-                self.piped_connections.push(piped_connection(pc.handle, pc.pipes));
+                self.piped_connections.push(piped_connection(pc.socket_id, pc.handle, pc.pipes));
                 continue;
             }
             if socket.state() == tcp::State::Closed {
@@ -1636,10 +1778,12 @@ fn main() {
     );
     const TOKEN_LISTENER: u64 = 0;
     const TOKEN_NIC: u64 = 1;
-    const TOKEN_TX_PIPE_BASE: u64 = 0x1000;
-    const TOKEN_RX_PIPE_BASE: u64 = 0x8000;
-    // Clear of the tx- and rx-pipe ranges by more than `MAX_PIPED_SLOTS`, and of a
-    // connection's own handle by more than `MAX_HANDLES` (4096,
+    // A piped connection's two pipes, by its socket id in the low word: an
+    // answer names the connection it was asked of and no place in a list,
+    // which a connection let go since would hand to another.
+    const TOKEN_SEND_PIPE: u64 = 1 << 32;
+    const TOKEN_RECEIVE_PIPE: u64 = 2 << 32;
+    // Clear of a connection's own handle by more than `MAX_HANDLES` (4096,
     // `kernel/src/object/handle.rs`).
     const TOKEN_PENDING_BASE: u64 = 0x1_0000;
     // Clear of the pending range by the same margin.
@@ -1722,19 +1866,23 @@ fn main() {
         poller.watch(&acceptor, READABLE, TOKEN_LISTENER);
         poller.watch(device.nic.claim(), READABLE, TOKEN_NIC);
 
-        // The client's bytes to send, and room in a receive pipe that is
-        // holding the peer's back: either is a pass's worth of work.
-        for (i, conn) in daemon.piped_connections.iter().enumerate() {
-            // Only while the socket can take them: a pipe holding bytes is
-            // readable until read, so its watch would complete on every pass
-            // while the peer's window is shut. The ACK that makes room wakes
-            // the NIC.
+        // The client's bytes to send, room in a receive pipe that is holding
+        // the peer's back, and the client letting go of either end: each is a
+        // pass's worth of work.
+        for conn in daemon.piped_connections.iter() {
+            // Readable only while the socket can take the bytes: a pipe
+            // holding some is readable until read, so its watch would complete
+            // on every pass while the peer's window is shut. The ACK that makes
+            // room wakes the NIC. Its writer's leaving is asked until answered,
+            // for the same reason: it stays so.
             let room = send_room(socket_set.get::<tcp::Socket>(conn.handle));
-            if let (true, Some(pipe)) = (room, &conn.tx_read) {
-                poller.watch(pipe, READABLE, TOKEN_TX_PIPE_BASE + i as u64);
+            let send = if room { READABLE } else { 0 } | if conn.writer_gone { 0 } else { OTHER_END_GONE };
+            if let (true, Some(pipe)) = (send != 0, &conn.tx_read) {
+                poller.watch(pipe, send, TOKEN_SEND_PIPE | u64::from(conn.socket_id));
             }
-            if let (true, Some(pipe)) = (conn.held, &conn.rx_write) {
-                poller.watch(pipe, WRITABLE, TOKEN_RX_PIPE_BASE + i as u64);
+            if let Some(pipe) = &conn.rx_write {
+                let room = if conn.held { WRITABLE } else { 0 };
+                poller.watch(pipe, room | OTHER_END_GONE, TOKEN_RECEIVE_PIPE | u64::from(conn.socket_id));
             }
         }
 
@@ -1758,6 +1906,10 @@ fn main() {
             Some(left) => timeout.min(left.as_nanos() as u64),
             None => timeout,
         };
+        let timeout = match daemon.ownerless_wake_in() {
+            Some(left) => timeout.min(left.as_nanos() as u64),
+            None => timeout,
+        };
         // The probe's window is a wake of its own: an idle machine would
         // otherwise sleep through the moment it owes its answer.
         let timeout = match &report {
@@ -1781,7 +1933,11 @@ fn main() {
         };
 
         let mut ready: Vec<u64> = Vec::new();
-        poller.wait(1, timeout, |token| ready.push(token));
+        poller.wait_answers(1, timeout, |token, answer| match token & !u64::from(u32::MAX) {
+            TOKEN_SEND_PIPE => daemon.pipe_answered(&mut socket_set, token as u32, true, answer),
+            TOKEN_RECEIVE_PIPE => daemon.pipe_answered(&mut socket_set, token as u32, false, answer),
+            _ => ready.push(token),
+        });
 
         // A handshake that never completes is why this deadline exists, and the
         // sweep has to happen on a pass that found nothing ready too —