diff --git a/CHANGELOG.md b/CHANGELOG.md
index 48bf506e..4666e42a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,10 @@
 
 ### Fixed
 
+- **The embedded terminal no longer leaks a file descriptor, reader thread, and shell process on exit.** When the terminal client disconnected or the shell exited, the PTY fd, its reader thread, and the shell process were left dangling — accumulating over a long uptime (part of the #4633 long-uptime-crash family). Teardown now closes the fd once, stops the reader thread, and reaps the shell on both the client-exit and shell-exit paths. Thanks @ai-ag2026. (#5835, #4633)
+
+- **Atomic JSON writers now `fsync` before rename, and the discoverability temp write is thread-safe.** The OAuth, passkeys, and session-discoverability stores wrote via temp-file → `os.replace` but without an `fsync`, so a crash right after the rename could still surface a partially-flushed file; they now flush → `fsync` → replace (matching the settings/config atomic-write hardening). The discoverability temp file also now uses a thread-specific name so concurrent writers can't collide. Thanks @ai-ag2026. (#5843)
+
 - **The dashboard status poll now pauses while the browser tab is hidden.** The Insights/dashboard status poll kept firing on a backgrounded tab; it now skips unforced polls while `document.hidden` and does one catch-up fetch on re-show (sibling of the hidden-tab session-poll pause). Forced init/save/resume fetches are unaffected. Thanks @ai-ag2026. (#5840)
 
 - **A slow or stuck client can no longer pin the session-events invalidation stream.** The session-events SSE stream now arms a write deadline after headers and before subscription, so a client that stops reading is torn down (and unsubscribed in `finally`) instead of holding the stream open indefinitely. A healthy slow-but-alive client is unaffected. Thanks @ai-ag2026. (#5841)
diff --git a/api/oauth.py b/api/oauth.py
index 8e928cb9..b9f6495b 100644
--- a/api/oauth.py
+++ b/api/oauth.py
@@ -191,7 +191,10 @@ def _write_auth_json(data: dict[str, Any], auth_path: Path | None = None) -> Pat
     path.parent.mkdir(parents=True, exist_ok=True)
     tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
     try:
-        tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+        with open(tmp, "w", encoding="utf-8") as f:
+            f.write(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
+            f.flush()
+            os.fsync(f.fileno())  # durable before rename: no truncated auth.json on power loss
         try:
             tmp.chmod(0o600)
         except OSError as exc:
diff --git a/api/passkeys.py b/api/passkeys.py
index 1bdc800f..b5bead77 100644
--- a/api/passkeys.py
+++ b/api/passkeys.py
@@ -72,6 +72,8 @@ def _atomic_write_json(path: Path, payload: Any) -> None:
     try:
         with os.fdopen(fd, "w", encoding="utf-8") as f:
             json.dump(payload, f, indent=2, sort_keys=True)
+            f.flush()
+            os.fsync(f.fileno())  # durable before rename: no zero-length passkey file on power loss
         os.chmod(tmp, 0o600)
         os.replace(tmp, path)
     except Exception:
diff --git a/api/session_discoverability.py b/api/session_discoverability.py
index cb3de719..630ccdf6 100644
--- a/api/session_discoverability.py
+++ b/api/session_discoverability.py
@@ -13,6 +13,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
+import threading
 import shutil
 import sqlite3
 from collections import Counter
@@ -387,9 +388,27 @@ def audit_session_discoverability(
 
 
 def _atomic_write_json(path: Path, payload) -> None:
-    tmp = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
-    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
-    os.replace(tmp, path)
+    # Temp name includes the thread id, not just the pid: WebUI runs a
+    # ThreadingHTTPServer, so two request threads reconciling the same _index.json
+    # would otherwise collide on one `<name>.tmp.<pid>` path and truncate each
+    # other's temp before its os.replace. fsync before the rename so a power loss
+    # can't leave a zero-length/garbage file where the durable JSON should be; a
+    # finally drops the temp if anything fails. Mode matches the prior write_text
+    # (umask default) since the plain open() respects the umask.
+    text = json.dumps(payload, ensure_ascii=False, indent=2)
+    tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp")
+    try:
+        with open(tmp, "w", encoding="utf-8") as f:
+            f.write(text)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp, path)
+    except BaseException:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        raise
 
 
 def _backup_file(path: Path, backup_dir: Path, backed_up: dict[Path, str]) -> str | None:
@@ -482,7 +501,7 @@ def _materialize_sidecar_from_state_db(session_dir: Path, state_db_path: Path |
     payload = _state_db_row_to_sidecar(row)
     _backup_file(state_db_path, backup_dir, backed_up)
     session_dir.mkdir(parents=True, exist_ok=True)
-    tmp = target.with_suffix(target.suffix + f".tmp.{os.getpid()}")
+    tmp = target.with_suffix(target.suffix + f".tmp.{os.getpid()}.{threading.get_ident()}")
     tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
     try:
         os.link(str(tmp), str(target))
diff --git a/api/terminal.py b/api/terminal.py
index 18c1fedb..44e4a12a 100644
--- a/api/terminal.py
+++ b/api/terminal.py
@@ -64,11 +64,23 @@ class TerminalSession:
     output: queue.Queue = field(default_factory=lambda: queue.Queue(maxsize=2000))
     closed: threading.Event = field(default_factory=threading.Event)
     reader: threading.Thread | None = None
+    # Serializes fd-touching ops (os.write, resize ioctl) against os.close, so a
+    # write can never land on a master_fd that was closed and whose number was
+    # already recycled by a concurrent openpty — that would inject the user's
+    # keystrokes into a foreign fd. Holders re-check ``closed`` under this lock
+    # and bail if the terminal has been torn down.
+    io_lock: threading.Lock = field(default_factory=threading.Lock)
+    # Wall-clock of the last input written or output produced. Drives which
+    # terminal the cap evicts first (least-recently-active): a shell abandoned
+    # at its prompt has neither, so it sorts oldest and is evicted before an
+    # actively used one.
+    last_activity: float = field(default_factory=time.time)
 
     def is_alive(self) -> bool:
         return not self.closed.is_set() and self.proc.poll() is None
 
     def put_output(self, event: str, payload: dict) -> None:
+        self.last_activity = time.time()
         try:
             self.output.put_nowait((event, payload))
         except queue.Full:
@@ -85,6 +97,14 @@ class TerminalSession:
 
 _TERMINALS: dict[str, TerminalSession] = {}
 _LOCK = threading.RLock()
+# Hard cap on concurrently live embedded terminals. Each holds a shell process,
+# a pty master fd, and a reader thread; a client that drops its output stream
+# without POSTing /api/terminal/close (tab close, crash, network drop) leaves
+# the shell running (no PDEATHSIG — see the note below), so without a ceiling
+# these accumulate over a long uptime toward fd/thread exhaustion (#4633). The
+# cap evicts the least-recently-active terminal to make room. Generous enough
+# that real interactive use never trips it.
+_MAX_TERMINALS = 32
 _spawn_queue: queue.Queue = queue.Queue()
 _spawn_supervisor_started = False
 _spawn_supervisor_lock = threading.Lock()
@@ -278,15 +298,27 @@ def _reader_loop(term: TerminalSession) -> None:
         code = term.proc.poll()
         _reap_terminal_descendants(term.proc.pid)
         term.put_output("terminal_closed", {"exit_code": code})
+        # The shell has exited (or its pty broke): retire the session so its
+        # master fd and _TERMINALS entry are released. Previously only an
+        # explicit close_terminal() / restart / atexit did this, so a shell that
+        # exited on its own (user typed `exit`, process died) leaked its master
+        # fd and dict entry for the rest of the WebUI's uptime. ``expected=term``
+        # makes this a no-op if a restart already replaced the entry with a new
+        # terminal for the same session id.
+        close_terminal(term.session_id, expected=term)
 
 
 def _set_size(term: TerminalSession, rows: int, cols: int) -> None:
     term.rows = max(8, min(int(rows or term.rows or 24), 80))
     term.cols = max(20, min(int(cols or term.cols or 80), 240))
-    try:
-        fcntl.ioctl(term.master_fd, termios.TIOCSWINSZ, _winsize(term.rows, term.cols))
-    except OSError:
-        pass
+    # The ioctl touches master_fd, so guard it against a concurrent close (and
+    # the fd-number recycling that can follow) with the same io_lock as writes.
+    with term.io_lock:
+        if not term.closed.is_set():
+            try:
+                fcntl.ioctl(term.master_fd, termios.TIOCSWINSZ, _winsize(term.rows, term.cols))
+            except OSError:
+                pass
     try:
         if term.proc.poll() is None:
             os.killpg(term.proc.pid, signal.SIGWINCH)
@@ -294,6 +326,43 @@ def _set_size(term: TerminalSession, rows: int, cols: int) -> None:
         pass
 
 
+def _enforce_terminal_cap(*, exclude_sid: str | None = None) -> None:
+    """Evict terminals until there is room under ``_MAX_TERMINALS``.
+
+    Picks a victim under the lock but closes it *outside* the lock (close_terminal
+    may spend up to a couple of seconds killing/​waiting on the shell). Prefers a
+    dead-process terminal, else the least-recently-active one — an abandoned
+    shell idle at its prompt sorts oldest and goes first. ``exclude_sid`` is the
+    session about to reuse/replace its own entry, so it never evicts itself.
+    """
+    if not _TERMINAL_SUPPORTED:
+        return
+    # Bounded loop: at most the current population; guards against a pathological
+    # spin if close_terminal somehow can't remove an entry.
+    for _ in range(_MAX_TERMINALS + 1):
+        victim_sid = None
+        victim_term = None
+        with _LOCK:
+            # Reuse/restart of an existing sid replaces in place — no growth.
+            if exclude_sid in _TERMINALS:
+                return
+            if len(_TERMINALS) < _MAX_TERMINALS:
+                return
+            candidates = [
+                (sid, term) for sid, term in _TERMINALS.items() if sid != exclude_sid
+            ]
+            if not candidates:
+                return
+            dead = [(sid, term) for sid, term in candidates if not term.is_alive()]
+            victim_sid, victim_term = (
+                dead[0] if dead else min(candidates, key=lambda kv: kv[1].last_activity)
+            )
+        # ``expected=victim_term`` so that if this sid was restarted/replaced in
+        # the gap between picking it and closing it, we don't tear down the new
+        # (possibly active) terminal — symmetric with the reader-loop retire.
+        close_terminal(victim_sid, expected=victim_term)
+
+
 def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int = 80, restart: bool = False) -> TerminalSession:
     """Start or return the embedded terminal for a WebUI session."""
     if not _TERMINAL_SUPPORTED:
@@ -305,6 +374,12 @@ def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int =
     if not Path(cwd).is_dir():
         raise ValueError("workspace is not a directory")
 
+    # Enforce the cap before spawning. Done outside the main lock below (the
+    # eviction's process teardown must not run while _LOCK is held for the whole
+    # spawn), and skipped for a same-sid reuse/restart, which replaces rather
+    # than adds an entry.
+    _enforce_terminal_cap(exclude_sid=sid)
+
     with _LOCK:
         current = _TERMINALS.get(sid)
         if current and current.is_alive() and not restart and current.workspace == cwd:
@@ -402,7 +477,14 @@ def write_terminal(session_id: str, data: str) -> None:
     term = get_terminal(session_id)
     if not term or not term.is_alive():
         raise KeyError("terminal not running")
-    os.write(term.master_fd, str(data or "").encode("utf-8", errors="replace"))
+    # Re-check ``closed`` under io_lock and write while holding it, so the fd
+    # can't be closed (and its number recycled by another openpty) between the
+    # check and the write — which would inject this input into a foreign fd.
+    with term.io_lock:
+        if term.closed.is_set():
+            raise KeyError("terminal not running")
+        os.write(term.master_fd, str(data or "").encode("utf-8", errors="replace"))
+    term.last_activity = time.time()
 
 
 def resize_terminal(session_id: str, rows: int, cols: int) -> None:
@@ -414,11 +496,21 @@ def resize_terminal(session_id: str, rows: int, cols: int) -> None:
     _set_size(term, rows, cols)
 
 
-def close_terminal(session_id: str) -> bool:
+def close_terminal(session_id: str, *, expected: TerminalSession | None = None) -> bool:
+    """Tear down the terminal for *session_id*: kill the shell, close the pty
+    master fd, reap descendants, and drop the ``_TERMINALS`` entry.
+
+    ``expected`` guards the retire-from-reader-loop path: only act if the live
+    entry is still that exact terminal, so an old reader thread finishing after
+    a restart cannot tear down the *new* terminal that replaced it (the old
+    one's fd was already closed by the restart's own close_terminal call).
+    """
     if not _TERMINAL_SUPPORTED:
         return False
     sid = str(session_id or "")
     with _LOCK:
+        if expected is not None and _TERMINALS.get(sid) is not expected:
+            return False
         term = _TERMINALS.pop(sid, None)
     if not term:
         return False
@@ -441,10 +533,13 @@ def close_terminal(session_id: str) -> bool:
                 except (subprocess.TimeoutExpired, ProcessLookupError):
                     pass
     finally:
-        try:
-            os.close(term.master_fd)
-        except OSError:
-            pass
+        # ``closed`` is already set above, so a writer/​resizer blocked on io_lock
+        # will see it and bail rather than touch the fd we are about to close.
+        with term.io_lock:
+            try:
+                os.close(term.master_fd)
+            except OSError:
+                pass
         _reap_terminal_descendants(term.proc.pid)
     return True
 
diff --git a/tests/test_atomic_writer_fsync.py b/tests/test_atomic_writer_fsync.py
new file mode 100644
index 00000000..b34bf6c7
--- /dev/null
+++ b/tests/test_atomic_writer_fsync.py
@@ -0,0 +1,105 @@
+"""Regression tests — atomic JSON writers fsync and use collision-safe temps.
+
+Three writers wrote to a temp file and os.replace'd it into place but never
+fsynced, so a power loss between the write and the physical flush could leave a
+zero-length/garbage file where durable JSON should be:
+  - session_discoverability._atomic_write_json (also used a pid-only temp name,
+    which collides between two request threads of the same ThreadingHTTPServer
+    process writing the same _index.json),
+  - passkeys._atomic_write_json (credentials),
+  - oauth._write_auth_json (access/refresh tokens).
+
+Each now fsyncs before the rename; session_discoverability additionally puts the
+thread id in the temp name and cleans it up on failure. These pin the fsync, the
+per-thread temp name, and that content/permissions are unchanged.
+"""
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+
+def _spy_fsync(monkeypatch, module):
+    calls = []
+    real = os.fsync
+    monkeypatch.setattr(module.os, "fsync", lambda fd: (calls.append(fd), real(fd))[1])
+    return calls
+
+
+def test_session_discoverability_writer_fsyncs_and_roundtrips(tmp_path, monkeypatch):
+    import api.session_discoverability as sd
+
+    calls = _spy_fsync(monkeypatch, sd)
+    path = tmp_path / "_index.json"
+    payload = [{"session_id": "s1", "title": "Grüße 🌍"}]
+
+    sd._atomic_write_json(path, payload)
+
+    assert calls, "atomic write did not fsync before rename"
+    assert json.loads(path.read_text(encoding="utf-8")) == payload
+    assert [p.name for p in tmp_path.iterdir()] == ["_index.json"]  # no temp debris
+
+
+def test_session_discoverability_temp_name_is_thread_unique(monkeypatch):
+    """Two threads (same pid) must not collide on one temp path."""
+    import api.session_discoverability as sd
+
+    names = set()
+    real_replace = os.replace
+
+    def capture(src, dst):
+        names.add(Path(src).name)
+        return real_replace(src, dst)
+
+    monkeypatch.setattr(sd.os, "replace", capture)
+
+    import tempfile as _tf
+    d = Path(_tf.mkdtemp())
+    idents = iter([1111, 2222])
+    monkeypatch.setattr(sd.threading, "get_ident", lambda: next(idents))
+    sd._atomic_write_json(d / "_index.json", [{"a": 1}])
+    sd._atomic_write_json(d / "_index.json", [{"a": 2}])
+
+    assert len(names) == 2, f"temp names collided across threads: {names}"
+
+
+def test_session_discoverability_failed_write_leaves_no_debris(tmp_path, monkeypatch):
+    import api.session_discoverability as sd
+
+    original = tmp_path / "_index.json"
+    original.write_text('["keep"]', encoding="utf-8")
+
+    def boom(src, dst):
+        raise RuntimeError("simulated crash before rename")
+
+    monkeypatch.setattr(sd.os, "replace", boom)
+    with pytest.raises(RuntimeError):
+        sd._atomic_write_json(original, [{"new": True}])
+
+    assert original.read_text(encoding="utf-8") == '["keep"]'  # original intact
+    assert [p.name for p in tmp_path.iterdir()] == ["_index.json"]  # temp cleaned up
+
+
+def test_passkeys_writer_fsyncs_and_keeps_0600(tmp_path, monkeypatch):
+    import api.passkeys as passkeys
+
+    calls = _spy_fsync(monkeypatch, passkeys)
+    path = tmp_path / "passkeys.json"
+    passkeys._atomic_write_json(path, [{"id": "cred-1"}])
+
+    assert calls, "passkeys write did not fsync"
+    assert json.loads(path.read_text(encoding="utf-8")) == [{"id": "cred-1"}]
+    assert (os.stat(path).st_mode & 0o777) == 0o600  # owner-only preserved
+
+
+def test_oauth_writer_fsyncs_and_keeps_0600(tmp_path, monkeypatch):
+    import api.oauth as oauth
+
+    calls = _spy_fsync(monkeypatch, oauth)
+    path = tmp_path / "auth.json"
+    oauth._write_auth_json({"access_token": "x"}, auth_path=path)
+
+    assert calls, "oauth auth.json write did not fsync"
+    assert json.loads(path.read_text(encoding="utf-8")) == {"access_token": "x"}
+    assert (os.stat(path).st_mode & 0o777) == 0o600  # owner-only preserved
diff --git a/tests/test_terminal_fd_leak.py b/tests/test_terminal_fd_leak.py
new file mode 100644
index 00000000..41604d84
--- /dev/null
+++ b/tests/test_terminal_fd_leak.py
@@ -0,0 +1,222 @@
+"""Regression tests for the embedded-terminal resource leak (#4633).
+
+Two leaks are closed here:
+
+1. **fd + dict entry on shell exit.** When a shell exited on its own (user typed
+   `exit`, the process died), `_reader_loop` broke out of its loop but only set
+   `closed` and emitted `terminal_closed`; it never closed the pty `master_fd`
+   nor removed the `_TERMINALS` entry. Only an explicit `close_terminal` /
+   restart / atexit did that, so a self-exited shell leaked its fd and dict
+   entry for the rest of the WebUI's uptime. The reader loop now retires the
+   session (identity-guarded so a restart's replacement is never touched).
+
+2. **Unbounded accumulation of abandoned terminals.** A client that drops its
+   output stream without POSTing /api/terminal/close (tab close, crash, network
+   drop) leaves the shell running (no PDEATHSIG, by design). Without a ceiling
+   these pile up toward fd/thread exhaustion. `_MAX_TERMINALS` now caps the live
+   population, evicting the least-recently-active terminal to make room.
+
+These use fakes for the shell/pty so they run without spawning real processes;
+the real-shell path is covered by test_terminal_linux_lifecycle.
+"""
+import os
+
+import pytest
+
+if os.name != "posix":
+    pytest.skip("terminal tests require POSIX terminal support", allow_module_level=True)
+
+import api.terminal as terminal
+
+
+class _FakeProc:
+    """A shell process fake. ``alive`` controls poll()."""
+
+    def __init__(self, pid=424242, alive=True):
+        self.pid = pid
+        self._alive = alive
+        self.wait_calls = []
+
+    def poll(self):
+        return None if self._alive else 0
+
+    def wait(self, timeout=None):
+        self.wait_calls.append(timeout)
+        return 0
+
+
+def _make_registered_term(monkeypatch, sid, *, alive=True, last_activity=0.0, real_fd=True):
+    """Build a TerminalSession, register it in _TERMINALS, no reader thread."""
+    if real_fd:
+        r, w = os.pipe()
+        os.close(w)  # leave a real, closable fd as master_fd
+        master_fd = r
+    else:
+        master_fd = -1
+    term = terminal.TerminalSession(
+        session_id=sid,
+        workspace="/tmp",
+        proc=_FakeProc(alive=alive),
+        master_fd=master_fd,
+        last_activity=last_activity,
+    )
+    with terminal._LOCK:
+        terminal._TERMINALS[sid] = term
+    return term
+
+
+@pytest.fixture(autouse=True)
+def _clean_terminals(monkeypatch):
+    # Never kill real process groups in these unit tests.
+    monkeypatch.setattr(terminal.os, "killpg", lambda *a, **k: None)
+    yield
+    with terminal._LOCK:
+        sids = list(terminal._TERMINALS)
+    for sid in sids:
+        try:
+            terminal.close_terminal(sid)
+        except Exception:
+            pass
+
+
+# ── Leak 1: reader loop retires the session on shell exit ────────────────────
+
+def test_reader_loop_retires_session_and_closes_fd(monkeypatch):
+    sid = "leak-reader-exit"
+    term = _make_registered_term(monkeypatch, sid, alive=False)
+    fd = term.master_fd
+
+    # Run the reader loop directly: a dead proc makes it exit its first iteration.
+    terminal._reader_loop(term)
+
+    with terminal._LOCK:
+        assert sid not in terminal._TERMINALS, "entry not retired on shell exit"
+    with pytest.raises(OSError):
+        os.fstat(fd)  # master_fd was closed — no leak
+
+
+def test_reader_loop_retire_is_identity_guarded_against_restart(monkeypatch):
+    """An old reader thread finishing after a restart must not tear down the
+    new terminal that replaced its session id."""
+    sid = "leak-restart-race"
+    old = _make_registered_term(monkeypatch, sid, alive=False)
+    # Simulate a restart: a NEW terminal now occupies the same sid.
+    new = _make_registered_term(monkeypatch, sid, alive=True)
+    assert terminal._TERMINALS[sid] is new
+
+    # The OLD reader loop finishes now.
+    terminal._reader_loop(old)
+
+    # The new terminal must still be registered and its fd open.
+    assert terminal._TERMINALS.get(sid) is new
+    os.fstat(new.master_fd)  # not closed
+
+
+# ── Leak 2: _MAX_TERMINALS cap evicts the least-recently-active terminal ──────
+
+def test_cap_evicts_least_recently_active(monkeypatch):
+    monkeypatch.setattr(terminal, "_MAX_TERMINALS", 3)
+    # Fill to the cap with alive terminals of increasing activity.
+    terms = {}
+    for i in range(3):
+        terms[i] = _make_registered_term(
+            monkeypatch, f"cap-{i}", alive=True, last_activity=100.0 + i
+        )
+    # cap-0 is least-recently-active. Enforcing the cap for a NEW sid evicts it.
+    terminal._enforce_terminal_cap(exclude_sid="cap-new")
+
+    with terminal._LOCK:
+        live = set(terminal._TERMINALS)
+    assert "cap-0" not in live, "least-recently-active terminal was not evicted"
+    assert {"cap-1", "cap-2"} <= live
+    assert len(live) < 3  # room was made for the new terminal
+
+
+def test_cap_prefers_dead_terminals_for_eviction(monkeypatch):
+    monkeypatch.setattr(terminal, "_MAX_TERMINALS", 3)
+    # A dead terminal that is NOT the least-recently-active must still go first.
+    _make_registered_term(monkeypatch, "cap-dead", alive=False, last_activity=999.0)
+    _make_registered_term(monkeypatch, "cap-live-a", alive=True, last_activity=1.0)
+    _make_registered_term(monkeypatch, "cap-live-b", alive=True, last_activity=2.0)
+
+    terminal._enforce_terminal_cap(exclude_sid="cap-new")
+
+    with terminal._LOCK:
+        live = set(terminal._TERMINALS)
+    assert "cap-dead" not in live, "dead terminal not preferred for eviction"
+    assert {"cap-live-a", "cap-live-b"} <= live
+
+
+def test_cap_reuse_of_existing_sid_evicts_nothing(monkeypatch):
+    monkeypatch.setattr(terminal, "_MAX_TERMINALS", 2)
+    _make_registered_term(monkeypatch, "keep-a", alive=True, last_activity=1.0)
+    _make_registered_term(monkeypatch, "keep-b", alive=True, last_activity=2.0)
+
+    # Reusing an already-registered sid replaces in place — no growth, no evict.
+    terminal._enforce_terminal_cap(exclude_sid="keep-a")
+
+    with terminal._LOCK:
+        assert {"keep-a", "keep-b"} <= set(terminal._TERMINALS)
+
+
+# ── F1: writes/resizes are serialized against close (no fd-reuse injection) ───
+
+def test_write_after_close_raises_and_never_touches_fd(monkeypatch):
+    """A write racing a teardown must not reach os.write once the terminal is
+    closed — otherwise the recycled fd number could receive foreign input."""
+    sid = "io-write-after-close"
+    term = _make_registered_term(monkeypatch, sid, alive=True)
+
+    calls = []
+    monkeypatch.setattr(terminal.os, "write", lambda fd, data: calls.append(fd))
+
+    # Simulate the teardown having marked the terminal closed.
+    term.closed.set()
+    with pytest.raises(KeyError):
+        terminal.write_terminal(sid, "rm -rf /\n")
+    assert calls == [], "write reached os.write after close — fd-reuse risk"
+
+
+def test_close_acquires_io_lock_before_closing_fd(monkeypatch):
+    """close_terminal must take io_lock around os.close so a concurrent writer
+    holding io_lock finishes (or bails) first — proving the two are serialized."""
+    import threading
+
+    sid = "io-close-serialized"
+    term = _make_registered_term(monkeypatch, sid, alive=False)
+    fd = term.master_fd
+
+    closed_fd = []
+    monkeypatch.setattr(terminal.os, "close", lambda f: closed_fd.append(f))
+
+    # Hold io_lock, then kick off a close in another thread and confirm it blocks
+    # on the fd-close until we release.
+    with term.io_lock:
+        t = threading.Thread(target=lambda: terminal.close_terminal(sid, expected=term))
+        t.start()
+        t.join(timeout=0.3)
+        assert closed_fd == [], "close closed the fd without waiting for io_lock"
+    t.join(timeout=2.0)
+    assert closed_fd == [fd], "close did not close the fd after io_lock released"
+
+
+# ── F2: cap eviction is identity-guarded ─────────────────────────────────────
+
+def test_cap_eviction_uses_expected_guard(monkeypatch):
+    monkeypatch.setattr(terminal, "_MAX_TERMINALS", 1)
+    victim = _make_registered_term(monkeypatch, "cap-victim", alive=True, last_activity=1.0)
+
+    seen = {}
+
+    real_close = terminal.close_terminal
+
+    def _spy_close(sid, *, expected=None):
+        seen["sid"] = sid
+        seen["expected"] = expected
+        return real_close(sid, expected=expected)
+
+    monkeypatch.setattr(terminal, "close_terminal", _spy_close)
+    terminal._enforce_terminal_cap(exclude_sid="cap-new")
+
+    assert seen.get("sid") == "cap-victim"
+    assert seen.get("expected") is victim, "cap eviction must pass expected= for identity safety"
diff --git a/tests/test_terminal_linux_lifecycle.py b/tests/test_terminal_linux_lifecycle.py
index f0de9308..01495225 100644
--- a/tests/test_terminal_linux_lifecycle.py
+++ b/tests/test_terminal_linux_lifecycle.py
@@ -366,9 +366,16 @@ def test_terminal_supervisor_continues_after_mixed_popen_failures(monkeypatch, t
     term_2 = start_terminal(sid_ok_2, tmp_path, restart=True)
 
     try:
+        # Both successful spawns produced distinct, registered terminals. We
+        # assert on the returned objects rather than re-reading _TERMINALS[sid]:
+        # these FakeProcs never hold the pty slave open, so the master EOFs
+        # immediately and the reader loop now correctly retires the entry (the
+        # fd-leak fix) — a real, alive shell keeps its pty open and stays
+        # registered (covered by the real-shell lifecycle tests above).
         assert term_1 is not term_2
-        assert terminal._TERMINALS[sid_ok_1].proc is term_1.proc
-        assert terminal._TERMINALS[sid_ok_2].proc is term_2.proc
+        assert term_1.proc is not term_2.proc
+        assert isinstance(term_1, terminal.TerminalSession)
+        assert isinstance(term_2, terminal.TerminalSession)
         assert terminal._spawn_supervisor_thread is not None
         assert terminal._spawn_supervisor_thread.is_alive()
     finally: