diff --git a/CHANGELOG.md b/CHANGELOG.md
index e01b8b2..5e55af4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,67 @@
 # Changelog
 
+## [Unreleased]
+
+### Added
+
+- `ClientPubkey`: the rmcp server worker now injects the caller's Nostr public
+  key (hex) into every **real** inbound request's `extensions` typemap, so
+  tool/resource/prompt handlers can identify their caller via
+  `ctx.extensions.get::<ClientPubkey>()`. It is not injected for the transport's
+  own synthetic announcement/initialization drives (which carry the
+  `ANNOUNCEMENT_REQUEST_ID` sentinel as their pubkey), so handlers never observe
+  a bogus caller; oversized (CEP-22) requests carry a real pubkey and are
+  injected as usual.
+  This closes the parity gap with the TS adapter's `extra._meta.clientPubkey`, but
+  uses rmcp's typed extensions (local-only, never on the wire) instead of the
+  `_meta` field, so it is always on rather than opt-in. The inbound event id is
+  already reachable as the rmcp request id (`ctx.id`).
+- `InboundEvent`: the rmcp server worker now also injects the **full**
+  client-signed Nostr request event into `extensions`, reachable via
+  `ctx.extensions.get::<InboundEvent>()`. For gift-wrapped requests this is the
+  inner, signature-verified event (its `pubkey` matches `ClientPubkey` by
+  construction); for plaintext requests it is the outer event; for CEP-22
+  oversized requests it is the carrying `end` frame's event. This exposes
+  `id`, `pubkey`, `sig`, `tags`, … — notably `sig`, which the server cannot
+  reconstruct without the client's private key. Handlers that must bind a tool
+  call to / store / audit the publishing event (e.g. an MLS key-package
+  coordinator returning the publication event) no longer have to fabricate a
+  synthetic event. Injected only for real client requests; synthetic
+  transport-internal requests carry none (`get` returns `None`).
+- `IncomingRequest` gained an `event: Option<nostr_sdk::Event>` field carrying
+  the same event through the channel seam. The FFI mirrors (`FfiIncomingRequest`,
+  the UniFFI `IncomingRequest`) intentionally do not surface it yet (no FFI
+  consumer needs the raw event; mirroring `nostr_sdk::Event` + `Tags` is
+  non-trivial) — the omission is documented in place.
+
+### Fixed
+
+- fix(open-stream): abort server writers on silent client disconnect (CEP-41).
+  Server→client `OpenStreamWriter`s leaked when a client silently disappeared
+  (crash/sleep/network drop) without sending `abort`. CEP-41 mandates each peer
+  maintain an idle timeout and probe the other with `ping`/`pong`, but only the
+  reader session ran keepalive timers; a pure producer stream (e.g. a
+  subscription-style tool streaming to a client) was never probed, so a dead
+  client left the writer — and any upstream producer keyed on `is_active()` —
+  alive indefinitely. The writer now arms an idle window once it starts
+  streaming, the server keepalive sweep probes it (mirroring the existing reader
+  sweep, driven by `OpenStreamWriter::tick`), an inbound `pong` for the stream is
+  routed to the writer to clear the probe (`ack_probe`), and a missing `pong`
+  aborts with `"Probe timeout"` (flushing any deferred final response via the
+  existing `on_abort` hook) **and evicts the dead client's session** (CEP-41
+  "release local state", mirroring the TS `handleProbeTimeout`, firing the
+  `SessionStore` eviction callback). Per CEP-41 only inbound frames reset the idle
+  window — a successful `write()` against the relay is not liveness. Reuses the
+  reader `idle_timeout_ms` / `probe_timeout_ms` knobs (one idle/probe pair per
+  stream). This is the rs-sdk port of the TS SDK 0.13.8 fix.
+- fix(server): verify plaintext event signatures before trusting `event.pubkey`
+  for handler identity, the auth allowlist, and request correlation, mirroring
+  the gift-wrap arm. The default `RelayPool` verifies inbound signatures itself,
+  but `RelayPoolTrait` is public, so a custom pool that skips verification
+  (e.g. `MockRelayPool`) left caller identity dependent on an undocumented pool
+  assumption — and a forged pubkey could bypass the auth allowlist. This is the
+  rs-sdk half of the TS identity-forgery fix (ContextVM/sdk#64, #69).
+
 ## [0.2.0] - 2026-06-24
 
 ### Added
diff --git a/contextvm-ffi/src/types.rs b/contextvm-ffi/src/types.rs
index e82c9c4..3053245 100644
--- a/contextvm-ffi/src/types.rs
+++ b/contextvm-ffi/src/types.rs
@@ -45,6 +45,10 @@ pub struct FfiJsonRpcMessage {
 }
 
 /// An FFI-safe incoming request (server-side).
+///
+/// NOTE: the SDK's `IncomingRequest` also carries an `event:
+/// Option<nostr_sdk::Event>` (the full client-signed event), intentionally not
+/// mirrored here — see `uniffi_types::IncomingRequest` for the rationale.
 #[repr(C)]
 #[derive(Debug)]
 pub struct FfiIncomingRequest {
diff --git a/contextvm-ffi/src/uniffi_types.rs b/contextvm-ffi/src/uniffi_types.rs
index ce23fd8..f7ecf5d 100644
--- a/contextvm-ffi/src/uniffi_types.rs
+++ b/contextvm-ffi/src/uniffi_types.rs
@@ -53,6 +53,13 @@ pub struct JsonRpcMessage {
 }
 
 /// An incoming MCP request (server-side).
+///
+/// NOTE: the SDK's `contextvm_sdk::IncomingRequest` also carries an
+/// `event: Option<nostr_sdk::Event>` field (the full client-signed event). It
+/// is intentionally NOT mirrored here: no FFI consumer needs the raw Nostr
+/// event today, and `nostr_sdk::Event` (with its `Tags` collection) is
+/// non-trivial to expose across the C ABI + UniFFI. Add an `Event` mirror and a
+/// field here when a foreign consumer needs `sig` / event binding.
 #[derive(Debug, Clone, uniffi::Record)]
 pub struct IncomingRequest {
     pub message: JsonRpcMessage,
@@ -231,6 +238,8 @@ fn incoming_to_uniffi(req: &contextvm_sdk::IncomingRequest) -> IncomingRequest {
         client_pubkey: req.client_pubkey.clone(),
         event_id: req.event_id.clone(),
         is_encrypted: req.is_encrypted,
+        // req.event (the full client-signed Nostr event) is deliberately not
+        // forwarded — see the note on `IncomingRequest` above.
     }
 }
 
diff --git a/docs/server-transport.md b/docs/server-transport.md
index c54ed9a..6a8c636 100644
--- a/docs/server-transport.md
+++ b/docs/server-transport.md
@@ -166,6 +166,77 @@ after the stream closes. Open-stream is disabled by default; enable it with
 `with_open_stream(OpenStreamConfig::enabled())`. See
 [open-stream.md](open-stream.md) for a full example.
 
+## Identifying the calling client
+
+Every inbound request carries its caller's Nostr public key. The rmcp worker
+injects it into the request `extensions`, so a tool, resource, or prompt handler
+reads it from `ctx.extensions`:
+
+```rust
+use contextvm_sdk::transport::server::ClientPubkey;
+use rmcp::handler::server::wrapper::Parameters;
+use rmcp::model::{CallToolResult, Content, ErrorData};
+use rmcp::service::RequestContext;
+use rmcp::{tool, RoleServer};
+
+#[tool(description = "Echo the caller's pubkey")]
+async fn whoami(
+    &self,
+    ctx: RequestContext<RoleServer>,
+) -> Result<CallToolResult, ErrorData> {
+    let pk = ctx.extensions.get::<ClientPubkey>().map(|c| c.0.clone());
+    Ok(CallToolResult::success(vec![Content::text(pk.unwrap_or_default())]))
+}
+```
+
+`ClientPubkey` lives in `contextvm_sdk::transport::server` (also re-exported at
+the crate root). It is present on every real inbound request, so retrieving it
+never returns `None` for a genuine client call. This mirrors the TS adapter's
+`extra._meta.clientPubkey`, but via rmcp's typed extensions instead of the
+on-wire `_meta` field. The inbound Nostr event id is available separately as
+`ctx.id` (the worker rewrites the request id to the event id).
+
+## The inbound Nostr event
+
+When a handler needs the **full** client-signed request event — not just the
+pubkey — the worker also injects an `InboundEvent` into `extensions`. This
+exposes the event's `id`, `pubkey`, `sig`, `tags`, …, which matters when a
+handler must bind a tool call to the publishing event, store it for later
+return, or audit it. `sig` in particular is the client's Schnorr signature and
+cannot be reconstructed by the server (it does not hold the client's private
+key), so it has to be threaded through from ingest.
+
+```rust
+use contextvm_sdk::transport::server::InboundEvent;
+use rmcp::handler::server::wrapper::Parameters;
+use rmcp::model::{CallToolResult, Content, ErrorData};
+use rmcp::service::RequestContext;
+use rmcp::{tool, RoleServer};
+
+#[tool(description = "Return the inbound event's id + sig")]
+async fn audit(
+    &self,
+    ctx: RequestContext<RoleServer>,
+) -> Result<CallToolResult, ErrorData> {
+    match ctx.extensions.get::<InboundEvent>() {
+        Some(ev) => Ok(CallToolResult::success(vec![Content::text(format!(
+            "id={} sig={}",
+            ev.0.id.to_hex(),
+            ev.0.sig // Display renders hex
+        ))])),
+        None => Err(ErrorData::invalid_params("no inbound event", None)),
+    }
+}
+```
+
+For gift-wrapped requests this is the **inner**, signature-verified event (the
+same one whose `pubkey` is surfaced as `ClientPubkey`, so `ev.0.pubkey` and
+`ClientPubkey` agree by construction); for plaintext requests it is the outer
+request event. It is injected only for real client requests — synthetic
+transport-internal requests (announcement / initialization drives, CEP-22
+oversized reassembly) carry no event, so `get::<InboundEvent>()` returns `None`
+for them. Like `ClientPubkey`, this is local-only and never touches the wire.
+
 ## When to use this instead of the gateway
 
 Use this page's approach when you are writing a new Rust MCP server.
diff --git a/sdk b/sdk
index 572926d..2aaa38a 160000
--- a/sdk
+++ b/sdk
@@ -1 +1 @@
-Subproject commit 572926d78adb11ec488aaedd4a377ea2ea4070bd
+Subproject commit 2aaa38afac378aa69bb0c02dc15811dcc2af9ea9
diff --git a/src/lib.rs b/src/lib.rs
index 0cad02b..58aa375 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -83,8 +83,8 @@ pub use transport::discovery_tags::{DiscoveredPeerCapabilities, PeerCapabilities
 
 // ── Transport (server) ──────────────────────────────────────────────
 pub use transport::server::{
-    IncomingRequest, NostrServerTransport, NostrServerTransportConfig, RouteEntry,
-    ServerEventRouteStore, SessionSnapshot, SessionStore,
+    ClientPubkey, InboundEvent, IncomingRequest, NostrServerTransport, NostrServerTransportConfig,
+    RouteEntry, ServerEventRouteStore, SessionSnapshot, SessionStore,
 };
 
 // ── rmcp re-export ──────────────────────────────────────────────────
diff --git a/src/rmcp_transport/worker.rs b/src/rmcp_transport/worker.rs
index 27d7dab..bd939e0 100644
--- a/src/rmcp_transport/worker.rs
+++ b/src/rmcp_transport/worker.rs
@@ -7,7 +7,9 @@ use crate::core::constants::ANNOUNCEMENT_REQUEST_ID;
 use crate::core::error::Result;
 use crate::core::types::{JsonRpcMessage, JsonRpcNotification, JsonRpcRequest};
 use crate::transport::client::{NostrClientTransport, NostrClientTransportConfig};
-use crate::transport::server::{NostrServerTransport, NostrServerTransportConfig};
+use crate::transport::server::{
+    ClientPubkey, InboundEvent, NostrServerTransport, NostrServerTransportConfig,
+};
 use rmcp::model::GetExtensions;
 use rmcp::transport::worker::{Worker, WorkerContext, WorkerQuitReason};
 use std::collections::HashSet;
@@ -148,6 +150,7 @@ impl Worker for NostrServerWorker {
                         mut message,
                         event_id,
                         client_pubkey,
+                        event,
                         ..
                     } = incoming;
 
@@ -206,13 +209,37 @@ impl Worker for NostrServerWorker {
                     }
 
                     if let Some(mut rmcp_msg) = internal_to_rmcp_server_rx(&message) {
-                        // CEP-41: inject the open-stream writer into the
-                        // request's `extensions` typemap so the tool handler can
-                        // reach it via `ctx.extensions.get::<OpenStreamWriter>()`.
-                        // The rmcp service loop swaps these extensions straight into
-                        // the handler's `RequestContext` before dispatch. No-op when
-                        // open-stream is disabled or the request has no writer.
+                        // Inject caller identity + (CEP-41) the open-stream writer
+                        // into the request's `extensions` typemap so the handler can
+                        // reach them via `ctx.extensions.get::<T>()`. The rmcp service
+                        // loop swaps these extensions straight into the handler's
+                        // `RequestContext` before dispatch.
                         if let rmcp::model::JsonRpcMessage::Request(ref mut jr) = rmcp_msg {
+                            // Caller pubkey is relevant to every real request (auth,
+                            // payments, logging) and extensions are local-only. Skip
+                            // the transport's own announcement/init drives, whose
+                            // "pubkey" is the `ANNOUNCEMENT_REQUEST_ID` sentinel —
+                            // injecting it would hand handlers a bogus caller. This
+                            // mirrors the TS adapter's `extra._meta.clientPubkey`,
+                            // but via rmcp's typed extensions rather than the on-wire
+                            // `_meta` field (which is why the TS knob is opt-in but
+                            // this is not). Gate on the sentinel, not on `event`:
+                            // oversized requests carry a real pubkey with no event.
+                            if client_pubkey != ANNOUNCEMENT_REQUEST_ID {
+                                jr.request
+                                    .extensions_mut()
+                                    .insert(ClientPubkey(client_pubkey.clone()));
+                            }
+                            // Full inbound event (id + sig + pubkey + …) for
+                            // handlers that must bind to / store / audit the
+                            // publishing event. Only real client requests carry
+                            // one; synthetic transport requests have none, so
+                            // handlers see `None` via get() for those.
+                            if let Some(ev) = event {
+                                jr.request.extensions_mut().insert(InboundEvent(ev));
+                            }
+                            // CEP-41: no-op when open-stream is disabled or the
+                            // request has no writer.
                             if let Some(writer) =
                                 self.transport.get_open_stream_writer(&event_id)
                             {
diff --git a/src/transport/open_stream/writer.rs b/src/transport/open_stream/writer.rs
index 0c1b160..4c2f244 100644
--- a/src/transport/open_stream/writer.rs
+++ b/src/transport/open_stream/writer.rs
@@ -16,13 +16,14 @@
 //! server transport wiring lands.
 
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex as StdMutex};
+use std::time::{Duration, Instant};
 
 use futures::future::BoxFuture;
 use tokio::sync::Mutex;
 
 use super::frame::OpenStreamFrame;
-use super::session::PublishFrame;
+use super::session::{KeepaliveAction, PublishFrame};
 
 /// Lifecycle hook fired after a terminal `close` frame is published.
 ///
@@ -47,14 +48,47 @@ pub struct OpenStreamWriterOptions {
     pub on_close: Option<OnCloseHook>,
     /// Fired after a terminal `abort` frame is published.
     pub on_abort: Option<OnAbortHook>,
+    /// Sender-side keepalive idle window (CEP-41 §Timeout and Keepalive). `None`
+    /// disables keepalive (e.g. unit tests). When set, the writer arms the window
+    /// once it starts streaming and probes the peer with `ping` if no inbound
+    /// frame arrives within this duration; a missing `pong` aborts with
+    /// `"Probe timeout"`, closing a leak where a silently-disconnected client
+    /// left the writer (and its upstream producer) alive indefinitely.
+    pub idle_timeout: Option<Duration>,
+    /// Time the writer waits for a matching `pong` after probing before aborting.
+    /// Ignored when [`idle_timeout`](OpenStreamWriterOptions::idle_timeout) is `None`.
+    pub probe_timeout: Duration,
+}
+
+impl OpenStreamWriterOptions {
+    /// Enable sender-side keepalive with the given idle/probe windows.
+    pub fn with_keepalive(mut self, idle: Duration, probe: Duration) -> Self {
+        self.idle_timeout = Some(idle);
+        self.probe_timeout = probe;
+        self
+    }
+}
+
+/// Sender-side keepalive state (CEP-41), guarded by a `std::sync::Mutex` so the
+/// pure sync [`tick`](OpenStreamWriter::tick) can mutate it without awaiting.
+///
+/// Per CEP-41 only **inbound** frames reset
+/// [`last_activity`](Self::last_activity) — our own `chunk` publishes do not (a
+/// vanished client keeps accepting relay events, so a successful `write` is not
+/// liveness).
+struct WriterKeepalive {
+    /// Last time an inbound frame was observed. Armed at `start`.
+    last_activity: Instant,
+    /// Nonce of an in-flight probe awaiting a matching `pong`.
+    pending_probe_nonce: Option<String>,
+    /// When the in-flight probe times out → abort.
+    probe_deadline: Option<Instant>,
 }
 
 /// State mutated only under the op `Mutex` (serialized publishes).
 struct WriterOpState {
     /// Next `chunkIndex` to assign (touched only by `write`).
     chunk_index: u64,
-    /// Control-frame nonce counter (touched only by `ping`).
-    control_nonce: u64,
 }
 
 struct WriterInner {
@@ -68,11 +102,18 @@ struct WriterInner {
     /// Monotonic outer `progress`, shared so `abort` can mint one without the op
     /// lock. Frames use `fetch_add(1) + 1` → 1, 2, 3, …
     progress: AtomicU64,
+    /// Control-frame nonce counter shared by manual `ping` and keepalive probes
+    /// so nonces stay unique within the stream (`{token}:{n}`).
+    control_nonce: AtomicU64,
     /// Liveness flag, **outside** the op lock so `abort` never blocks on a stuck
     /// write. `false` once closed or aborted.
     active: AtomicBool,
     /// Whether the lazy `start` frame has been published.
     started: AtomicBool,
+    /// Sender-side keepalive config + state. Inert when `idle_timeout` is `None`.
+    idle_timeout: Option<Duration>,
+    probe_timeout: Duration,
+    keepalive: StdMutex<WriterKeepalive>,
 }
 
 /// Minimal CEP-41 producer/writer.
@@ -91,13 +132,18 @@ impl OpenStreamWriter {
                 publish_frame: options.publish_frame,
                 on_close: options.on_close,
                 on_abort: options.on_abort,
-                op: Mutex::new(WriterOpState {
-                    chunk_index: 0,
-                    control_nonce: 0,
-                }),
+                op: Mutex::new(WriterOpState { chunk_index: 0 }),
                 progress: AtomicU64::new(0),
+                control_nonce: AtomicU64::new(0),
                 active: AtomicBool::new(true),
                 started: AtomicBool::new(false),
+                idle_timeout: options.idle_timeout,
+                probe_timeout: options.probe_timeout,
+                keepalive: StdMutex::new(WriterKeepalive {
+                    last_activity: Instant::now(),
+                    pending_probe_nonce: None,
+                    probe_deadline: None,
+                }),
             }),
         }
     }
@@ -144,6 +190,12 @@ impl OpenStreamWriter {
         )?;
         (self.inner.publish_frame)(notification).await?;
         self.inner.started.store(true, Ordering::SeqCst);
+        // Arm the sender-side keepalive idle window (CEP-41): the stream is active
+        // from `start`, so the probe clock starts here. Only inbound frames reset
+        // it afterwards — our own chunk publishes are not liveness.
+        if self.inner.idle_timeout.is_some() {
+            self.inner.keepalive.lock().unwrap().last_activity = Instant::now();
+        }
         Ok(())
     }
 
@@ -171,12 +223,12 @@ impl OpenStreamWriter {
 
     /// Publish a keepalive `ping` carrying a fresh `{token}:{n}` nonce.
     pub async fn ping(&self) -> crate::Result<()> {
-        let mut op = self.inner.op.lock().await;
+        let _op = self.inner.op.lock().await;
         if !self.inner.active.load(Ordering::SeqCst) {
             return Ok(());
         }
-        op.control_nonce += 1;
-        let nonce = format!("{}:{}", self.inner.progress_token, op.control_nonce);
+        let n = self.inner.control_nonce.fetch_add(1, Ordering::SeqCst) + 1;
+        let nonce = format!("{}:{}", self.inner.progress_token, n);
         let progress = self.next_progress();
         let notification = OpenStreamFrame::Ping { nonce }.into_progress_notification(
             &self.inner.progress_token,
@@ -188,11 +240,18 @@ impl OpenStreamWriter {
     }
 
     /// Publish a `pong` echoing the peer's `ping` nonce.
+    ///
+    /// An inbound `ping` is proof of peer liveness (CEP-41: receipt of any valid
+    /// frame resets the idle window), so this refreshes the keepalive clock
+    /// before responding.
     pub async fn pong(&self, nonce: String) -> crate::Result<()> {
         let _op = self.inner.op.lock().await;
         if !self.inner.active.load(Ordering::SeqCst) {
             return Ok(());
         }
+        if self.inner.idle_timeout.is_some() {
+            self.inner.keepalive.lock().unwrap().last_activity = Instant::now();
+        }
         let progress = self.next_progress();
         let notification = OpenStreamFrame::Pong { nonce }.into_progress_notification(
             &self.inner.progress_token,
@@ -203,6 +262,99 @@ impl OpenStreamWriter {
         Ok(())
     }
 
+    // ── sender-side keepalive (CEP-41), driven by the transport sweep ───────
+
+    /// Pure keepalive transition (idle → `ping`, probe deadline → abort), mirroring
+    /// the reader-side
+    /// [`OpenStreamSession::tick`](super::session::OpenStreamSession::tick).
+    /// Driven by the owning transport's periodic sweep with `Instant::now()`.
+    ///
+    /// Returns [`KeepaliveAction::None`] when keepalive is disabled, the writer is
+    /// inactive, or it has not yet started (the idle window arms at `start`).
+    pub fn tick(&self, now: Instant) -> KeepaliveAction {
+        let Some(idle_timeout) = self.inner.idle_timeout else {
+            return KeepaliveAction::None;
+        };
+        if !self.inner.active.load(Ordering::SeqCst) || !self.inner.started.load(Ordering::SeqCst) {
+            return KeepaliveAction::None;
+        }
+        let mut ka = self.inner.keepalive.lock().unwrap();
+
+        // 1. A pending probe whose deadline passed → abort.
+        if let Some(deadline) = ka.probe_deadline {
+            if ka.pending_probe_nonce.is_some() && now >= deadline {
+                return KeepaliveAction::Abort("Probe timeout".to_string());
+            }
+        }
+
+        // 2. Idle past the threshold (and not already probing) → ping.
+        if ka.pending_probe_nonce.is_none()
+            && now.saturating_duration_since(ka.last_activity) >= idle_timeout
+        {
+            let n = self.inner.control_nonce.fetch_add(1, Ordering::SeqCst) + 1;
+            let nonce = format!("{}:{}", self.inner.progress_token, n);
+            ka.pending_probe_nonce = Some(nonce.clone());
+            ka.probe_deadline = Some(now + self.inner.probe_timeout);
+            return KeepaliveAction::SendPing(nonce);
+        }
+
+        KeepaliveAction::None
+    }
+
+    /// Publish a keepalive `ping` carrying a `nonce` already minted by [`tick`](Self::tick).
+    ///
+    /// Used by the transport sweep; unlike [`ping`](Self::ping) it does not mint a
+    /// new nonce (the probe recorded by `tick` must match the frame on the wire so
+    /// a matching `pong` clears it). Does not take the op lock, so a stuck app
+    /// `write` cannot block keepalive (mirroring the TS writer's queue bypass).
+    ///
+    /// Ceiling: by skipping the op lock, this probe's `progress` (shared atomic)
+    /// can interleave with an in-flight `write` — a probe may carry a lower
+    /// `progress` than a chunk published just before it. Harmless: relays reorder
+    /// regardless and control frames aren't `chunkIndex`-gap-checked. Re-serialize
+    /// under the op lock only if strict monotonic publish-order is ever required
+    /// (which would re-block keepalive behind a stuck `write`).
+    pub async fn send_probe(&self, nonce: String) -> crate::Result<()> {
+        if !self.inner.active.load(Ordering::SeqCst) {
+            return Ok(());
+        }
+        let progress = self.next_progress();
+        let notification = OpenStreamFrame::Ping { nonce }.into_progress_notification(
+            &self.inner.progress_token,
+            progress,
+            None,
+        )?;
+        (self.inner.publish_frame)(notification).await.map(|_| ())
+    }
+
+    /// Acknowledge an inbound `pong` matching the pending keepalive probe (CEP-41).
+    ///
+    /// A matching `pong` clears the probe and refreshes the idle window. A `pong`
+    /// whose nonce does not match the pending probe is **not** evidence of
+    /// liveness (CEP-41: "A `pong` with an unknown … nonce MUST NOT be treated as
+    /// evidence of liveness") and is ignored.
+    pub fn ack_probe(&self, nonce: &str) {
+        let mut ka = self.inner.keepalive.lock().unwrap();
+        if self.inner.active.load(Ordering::SeqCst)
+            && ka.pending_probe_nonce.as_deref() == Some(nonce)
+        {
+            ka.pending_probe_nonce = None;
+            ka.probe_deadline = None;
+            ka.last_activity = Instant::now();
+        }
+    }
+
+    /// Release writer resources without publishing a terminal frame — used on
+    /// transport teardown so the writer no longer drives keepalive. Idempotent.
+    ///
+    /// Only flips `active` to `false`: [`tick`](Self::tick) and
+    /// [`ack_probe`](Self::ack_probe) early-return on that flag, so an armed probe
+    /// becomes inert without needing to be cleared (matching `close`/`abort`,
+    /// which also leave keepalive state untouched once terminal).
+    pub fn dispose(&self) {
+        let _ = self.inner.active.swap(false, Ordering::SeqCst);
+    }
+
     /// Close the stream gracefully. Declares `lastChunkIndex` iff any chunks were
     /// written. Runs [`on_close`](OpenStreamWriterOptions::on_close) **after** the
     /// frame is published (even on publish failure); propagates the publish error.
@@ -309,6 +461,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         })
     }
 
@@ -321,6 +475,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         assert!(writer.is_active());
@@ -356,6 +512,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         writer.start().await.unwrap();
@@ -419,6 +577,8 @@ mod tests {
             content_type: None,
             on_close: Some(on_close),
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
         writer.close().await.unwrap();
         assert_eq!(frame_type(log.lock().unwrap().last().unwrap()), "close");
@@ -439,6 +599,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: Some(on_abort),
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
         abort_writer.abort(Some("done".to_string())).await.unwrap();
         let frames = log.lock().unwrap();
@@ -476,6 +638,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: Some(on_abort),
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         writer.abort(Some("ordered".to_string())).await.unwrap();
@@ -509,6 +673,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         assert!(writer.start().await.is_err());
@@ -540,6 +706,8 @@ mod tests {
             content_type: None,
             on_close: Some(on_close),
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         assert!(writer.close().await.is_err());
@@ -573,6 +741,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: Some(on_abort),
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         assert!(writer.abort(Some("cleanup".to_string())).await.is_err());
@@ -616,6 +786,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: Some(on_abort),
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         let writer2 = writer.clone();
@@ -659,6 +831,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         });
 
         let (a, b, c) = tokio::join!(
@@ -682,4 +856,163 @@ mod tests {
         assert_eq!(progress_of(&frames[3]), 4);
         assert_eq!(cvm(&frames[3])["lastChunkIndex"], 1);
     }
+
+    // ── sender-side keepalive (CEP-41) ──────────────────────────────────────
+
+    /// A writer with sender-side keepalive armed (`idle_timeout = Some(..)`).
+    /// Returns the frame log so the probe publish can be inspected.
+    fn keepalive_writer(idle_ms: u64, probe_ms: u64) -> (OpenStreamWriter, FrameLog) {
+        let log = frame_log();
+        let writer = OpenStreamWriter::new(OpenStreamWriterOptions {
+            progress_token: "tok-ka".to_string(),
+            publish_frame: recording_publisher(log.clone()),
+            content_type: None,
+            on_close: None,
+            on_abort: None,
+            idle_timeout: Some(Duration::from_millis(idle_ms)),
+            probe_timeout: Duration::from_millis(probe_ms),
+        });
+        (writer, log)
+    }
+
+    #[tokio::test]
+    async fn keepalive_disabled_writer_tick_is_always_none() {
+        // `writer_with` builds a keepalive-disabled writer (idle_timeout: None).
+        let log = frame_log();
+        let writer = writer_with(log);
+        writer.start().await.unwrap();
+
+        let t = Instant::now();
+        assert_eq!(writer.tick(t), KeepaliveAction::None);
+        assert_eq!(
+            writer.tick(t + Duration::from_secs(60)),
+            KeepaliveAction::None
+        );
+    }
+
+    #[tokio::test]
+    async fn tick_returns_none_before_start() {
+        let (writer, _) = keepalive_writer(10, 20);
+        // Keepalive arms at `start`; a never-started writer must not probe.
+        assert_eq!(writer.tick(Instant::now()), KeepaliveAction::None);
+    }
+
+    #[tokio::test]
+    async fn tick_idle_probes_then_probe_deadline_aborts() {
+        let (writer, _) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+        let t0 = Instant::now();
+
+        // Before the idle threshold: nothing.
+        assert_eq!(writer.tick(t0), KeepaliveAction::None);
+        // At the idle threshold: probe with a `{token}:{n}` nonce.
+        let nonce = match writer.tick(t0 + Duration::from_millis(11)) {
+            KeepaliveAction::SendPing(n) => n,
+            other => panic!("expected SendPing, got {other:?}"),
+        };
+        assert!(nonce.starts_with("tok-ka:"));
+        // Probe in flight, deadline not yet reached: nothing.
+        assert_eq!(
+            writer.tick(t0 + Duration::from_millis(15)),
+            KeepaliveAction::None
+        );
+        // Probe deadline reached with no pong: abort.
+        assert_eq!(
+            writer.tick(t0 + Duration::from_millis(31)),
+            KeepaliveAction::Abort("Probe timeout".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn matching_pong_clears_probe_and_rearms_idle() {
+        let (writer, _) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+        let t0 = Instant::now();
+        let nonce = match writer.tick(t0 + Duration::from_millis(11)) {
+            KeepaliveAction::SendPing(n) => n,
+            other => panic!("expected SendPing, got {other:?}"),
+        };
+
+        // A matching pong clears the probe and refreshes liveness.
+        writer.ack_probe(&nonce);
+
+        // At the *old* probe deadline the stream is NOT aborted: the probe was
+        // cleared and the idle window restarted from ack time.
+        let after_ack = Instant::now();
+        assert!(!matches!(
+            writer.tick(after_ack + Duration::from_millis(5)),
+            KeepaliveAction::Abort(_)
+        ));
+        // A fresh probe fires only once the new idle window elapses.
+        assert!(matches!(
+            writer.tick(after_ack + Duration::from_millis(12)),
+            KeepaliveAction::SendPing(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn unmatched_pong_does_not_clear_probe() {
+        let (writer, _) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+        let t0 = Instant::now();
+        let _ = writer.tick(t0 + Duration::from_millis(11));
+
+        // A pong with a non-matching nonce must not clear the pending probe
+        // (CEP-41: an unknown nonce is not evidence of liveness).
+        writer.ack_probe("tok-ka:999");
+
+        assert_eq!(
+            writer.tick(t0 + Duration::from_millis(31)),
+            KeepaliveAction::Abort("Probe timeout".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn inbound_ping_refreshes_idle_window() {
+        let (writer, _) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+
+        // Handling an inbound `ping` (pong) is liveness — refreshes the idle
+        // window so a tick shortly after does not probe.
+        writer.pong("peer-ping".to_string()).await.unwrap();
+        let after_pong = Instant::now();
+        assert_eq!(
+            writer.tick(after_pong + Duration::from_millis(5)),
+            KeepaliveAction::None
+        );
+        // Once the idle window elapses again, it probes.
+        assert!(matches!(
+            writer.tick(after_pong + Duration::from_millis(12)),
+            KeepaliveAction::SendPing(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn send_probe_publishes_a_ping_with_the_given_nonce() {
+        let (writer, log) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+        writer.send_probe("custom-nonce".to_string()).await.unwrap();
+
+        // `frame_types` takes the `log` lock itself; call it before acquiring
+        // `frames` so the two lock acquisitions don't overlap (std::sync::Mutex is
+        // not re-entrant).
+        assert_eq!(frame_types(&log), vec!["start", "ping"]);
+        let frames = log.lock().unwrap();
+        assert_eq!(cvm(&frames[1])["nonce"], "custom-nonce");
+    }
+
+    #[tokio::test]
+    async fn dispose_makes_writer_inactive_and_tick_inert() {
+        let (writer, _) = keepalive_writer(10, 20);
+        writer.start().await.unwrap();
+        assert!(writer.is_active());
+
+        writer.dispose();
+
+        assert!(!writer.is_active());
+        assert_eq!(
+            writer.tick(Instant::now() + Duration::from_secs(60)),
+            KeepaliveAction::None
+        );
+    }
 }
diff --git a/src/transport/server/announcement_manager.rs b/src/transport/server/announcement_manager.rs
index c680899..fe45292 100644
--- a/src/transport/server/announcement_manager.rs
+++ b/src/transport/server/announcement_manager.rs
@@ -744,6 +744,7 @@ impl AnnouncementManager {
                         client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(),
                         event_id: ANNOUNCEMENT_REQUEST_ID.to_string(),
                         is_encrypted: false,
+                        event: None,
                     });
                 }
 
@@ -815,6 +816,7 @@ async fn publish_public_announcements(
             client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(),
             event_id: ANNOUNCEMENT_REQUEST_ID.to_string(),
             is_encrypted: false,
+            event: None,
         })
         .is_err()
     {
@@ -865,6 +867,7 @@ async fn publish_public_announcements(
             client_pubkey: ANNOUNCEMENT_REQUEST_ID.to_string(),
             event_id: ANNOUNCEMENT_REQUEST_ID.to_string(),
             is_encrypted: false,
+            event: None,
         });
     }
 
diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs
index 4841be2..aa1c4cc 100644
--- a/src/transport/server/mod.rs
+++ b/src/transport/server/mod.rs
@@ -437,8 +437,76 @@ pub struct IncomingRequest {
     pub event_id: String,
     /// Whether the original message was encrypted.
     pub is_encrypted: bool,
+    /// The inbound (client-signed) Nostr event, if this request carried one.
+    ///
+    /// `Some` for real client requests: the inner, signature-verified gift-wrap
+    /// event for encrypted requests, the outer event for plaintext, or the
+    /// carrying frame event for CEP-22 oversized reassembly. `None` only for
+    /// requests the transport synthesizes itself (announcement /
+    /// initialization drives), which carry no real Nostr event. Handlers reach
+    /// it via [`InboundEvent`] in request
+    /// `extensions`; see that type's docs.
+    pub event: Option<Event>,
 }
 
+/// The Nostr public key (hex) of the client that issued the current request.
+///
+/// [`NostrServerWorker`](crate::rmcp_transport::NostrServerWorker) injects this
+/// into every inbound request's rmcp `extensions` typemap, so a tool, resource,
+/// or prompt handler can identify its caller:
+///
+/// ```ignore
+/// use contextvm_sdk::transport::server::ClientPubkey;
+/// use rmcp::service::RequestContext;
+/// use rmcp::RoleServer;
+///
+/// let caller = ctx.extensions.get::<ClientPubkey>().map(|c| c.0.clone());
+/// ```
+///
+/// The value is the raw hex pubkey (the same string as
+/// [`IncomingRequest::client_pubkey`]). It is present on every real inbound
+/// request, unlike `OpenStreamWriter`, which is injected only for open-stream
+/// `tools/call` calls. This is purely a local affordance — it is never
+/// serialized onto the wire. The inbound Nostr event id is available separately
+/// as the rmcp request id (`ctx.id`), which the worker rewrites to the event id.
+#[derive(Debug, Clone)]
+pub struct ClientPubkey(pub String);
+
+/// The inbound (client-signed) Nostr event for the current request.
+///
+/// Injected into a request's rmcp `extensions` typemap by
+/// [`NostrServerWorker`](crate::rmcp_transport::NostrServerWorker) alongside
+/// [`ClientPubkey`], so a tool, resource, or prompt handler can read the full
+/// signed event — its `id`, `pubkey`, `sig`, … — to bind a tool call to the
+/// publishing event, store it for later return, or audit it:
+///
+/// ```ignore
+/// use contextvm_sdk::transport::server::InboundEvent;
+/// use rmcp::service::RequestContext;
+/// use rmcp::RoleServer;
+///
+/// if let Some(ev) = ctx.extensions.get::<InboundEvent>() {
+///     let sig_hex = ev.0.sig.to_string(); // Schnorr signature, hex
+///     // …
+/// }
+/// ```
+///
+/// For gift-wrapped requests this is the **inner**, signature-verified event
+/// (the same one whose `pubkey` is surfaced as [`ClientPubkey`], so `ev.0.pubkey`
+/// and `ClientPubkey` agree by construction). For plaintext requests it is the
+/// outer request event. For CEP-22 oversized requests it is the carrying `end`
+/// frame's event (a `notifications/progress` event signed by the client whose
+/// `id` is the correlation id) — the request was chunked, so no single dedicated
+/// request event exists. It is injected only for real client requests; synthetic
+/// transport-internal requests (announcement / initialization drives) carry no
+/// event, so `extensions.get::<InboundEvent>()` returns `None` for them. This is
+/// purely a local affordance — it is never
+/// serialized onto the wire. The event id is also reachable as the rmcp request
+/// id (`ctx.id`); this type additionally exposes `sig`, which the server cannot
+/// reconstruct without the client's private key.
+#[derive(Debug, Clone)]
+pub struct InboundEvent(pub Event);
+
 impl NostrServerTransport {
     /// Create a new server transport.
     pub async fn new<T>(signer: T, config: NostrServerTransportConfig) -> Result<Self>
@@ -735,7 +803,9 @@ impl NostrServerTransport {
             }
             receivers.clear();
         }
-        self.open_stream.lock_slots().clear();
+        for (_, slot) in self.open_stream.lock_slots().drain() {
+            slot.writer.dispose();
+        }
         self.open_stream.lock_token_index().clear();
         Ok(())
     }
@@ -1549,6 +1619,7 @@ impl NostrServerTransport {
                         &relay_pool,
                         encryption_mode,
                         gift_wrap_mode,
+                        &sessions,
                     )
                     .await;
                     continue;
@@ -1585,107 +1656,124 @@ impl NostrServerTransport {
                     continue;
                 }
 
-                let (content, sender_pubkey, event_id, is_encrypted, inner_tags) = if is_gift_wrap {
-                    if encryption_mode == EncryptionMode::Disabled {
-                        tracing::warn!(
-                            target: LOG_TARGET,
-                            event_id = %event.id.to_hex(),
-                            sender_pubkey = %event.pubkey.to_hex(),
-                            "Received encrypted message but encryption is disabled"
-                        );
-                        continue;
-                    }
-                    {
-                        let guard = match seen_gift_wrap_ids.lock() {
-                            Ok(g) => g,
-                            Err(poisoned) => poisoned.into_inner(),
-                        };
-                        if guard.contains(&event.id) {
-                            tracing::debug!(
+                let (content, sender_pubkey, event_id, is_encrypted, inner_tags, inbound_event) =
+                    if is_gift_wrap {
+                        if encryption_mode == EncryptionMode::Disabled {
+                            tracing::warn!(
                                 target: LOG_TARGET,
                                 event_id = %event.id.to_hex(),
-                                "Skipping duplicate gift-wrap (outer id)"
+                                sender_pubkey = %event.pubkey.to_hex(),
+                                "Received encrypted message but encryption is disabled"
                             );
                             continue;
                         }
-                    }
-                    // Single-layer NIP-44 decrypt (matches JS/TS SDK)
-                    let signer = match relay_pool.signer().await {
-                        Ok(s) => s,
-                        Err(error) => {
-                            tracing::error!(
-                                target: LOG_TARGET,
-                                error = %error,
-                                "Failed to get signer"
-                            );
-                            continue;
+                        {
+                            let guard = match seen_gift_wrap_ids.lock() {
+                                Ok(g) => g,
+                                Err(poisoned) => poisoned.into_inner(),
+                            };
+                            if guard.contains(&event.id) {
+                                tracing::debug!(
+                                    target: LOG_TARGET,
+                                    event_id = %event.id.to_hex(),
+                                    "Skipping duplicate gift-wrap (outer id)"
+                                );
+                                continue;
+                            }
                         }
-                    };
-                    match encryption::decrypt_gift_wrap_single_layer(&signer, &event).await {
-                        Ok(decrypted_json) => {
-                            // The decrypted content is JSON of the inner signed event.
-                            // Use the INNER event's ID for correlation — the client
-                            // registers the inner event ID in its correlation store.
-                            match serde_json::from_str::<Event>(&decrypted_json) {
-                                Ok(inner) => {
-                                    if let Err(e) = inner.verify() {
-                                        tracing::warn!(
-                                            "Inner event signature verification failed: {e}"
+                        // Single-layer NIP-44 decrypt (matches JS/TS SDK)
+                        let signer = match relay_pool.signer().await {
+                            Ok(s) => s,
+                            Err(error) => {
+                                tracing::error!(
+                                    target: LOG_TARGET,
+                                    error = %error,
+                                    "Failed to get signer"
+                                );
+                                continue;
+                            }
+                        };
+                        match encryption::decrypt_gift_wrap_single_layer(&signer, &event).await {
+                            Ok(decrypted_json) => {
+                                // The decrypted content is JSON of the inner signed event.
+                                // Use the INNER event's ID for correlation — the client
+                                // registers the inner event ID in its correlation store.
+                                match serde_json::from_str::<Event>(&decrypted_json) {
+                                    Ok(inner) => {
+                                        if let Err(e) = inner.verify() {
+                                            tracing::warn!(
+                                                "Inner event signature verification failed: {e}"
+                                            );
+                                            continue;
+                                        }
+                                        {
+                                            let mut guard = match seen_gift_wrap_ids.lock() {
+                                                Ok(g) => g,
+                                                Err(poisoned) => poisoned.into_inner(),
+                                            };
+                                            guard.put(event.id, ());
+                                        }
+                                        let inbound = inner.clone();
+                                        let inner_tags: Vec<Tag> = inner.tags.to_vec();
+                                        (
+                                            inner.content,
+                                            inner.pubkey.to_hex(),
+                                            inner.id.to_hex(),
+                                            true,
+                                            inner_tags,
+                                            Some(inbound),
+                                        )
+                                    }
+                                    Err(error) => {
+                                        tracing::error!(
+                                            target: LOG_TARGET,
+                                            error = %error,
+                                            "Failed to parse inner event"
                                         );
                                         continue;
                                     }
-                                    {
-                                        let mut guard = match seen_gift_wrap_ids.lock() {
-                                            Ok(g) => g,
-                                            Err(poisoned) => poisoned.into_inner(),
-                                        };
-                                        guard.put(event.id, ());
-                                    }
-                                    let inner_tags: Vec<Tag> = inner.tags.to_vec();
-                                    (
-                                        inner.content,
-                                        inner.pubkey.to_hex(),
-                                        inner.id.to_hex(),
-                                        true,
-                                        inner_tags,
-                                    )
-                                }
-                                Err(error) => {
-                                    tracing::error!(
-                                        target: LOG_TARGET,
-                                        error = %error,
-                                        "Failed to parse inner event"
-                                    );
-                                    continue;
                                 }
                             }
+                            Err(error) => {
+                                tracing::error!(
+                                    target: LOG_TARGET,
+                                    error = %error,
+                                    "Failed to decrypt"
+                                );
+                                continue;
+                            }
                         }
-                        Err(error) => {
-                            tracing::error!(
+                    } else {
+                        if encryption_mode == EncryptionMode::Required {
+                            tracing::warn!(
                                 target: LOG_TARGET,
-                                error = %error,
-                                "Failed to decrypt"
+                                sender_pubkey = %event.pubkey.to_hex(),
+                                "Received unencrypted message but encryption is required"
                             );
                             continue;
                         }
-                    }
-                } else {
-                    if encryption_mode == EncryptionMode::Required {
-                        tracing::warn!(
-                            target: LOG_TARGET,
-                            sender_pubkey = %event.pubkey.to_hex(),
-                            "Received unencrypted message but encryption is required"
-                        );
-                        continue;
-                    }
-                    (
-                        event.content.clone(),
-                        event.pubkey.to_hex(),
-                        event.id.to_hex(),
-                        false,
-                        event.tags.to_vec(),
-                    )
-                };
+                        // Verify the signature (and that `id` matches content) before
+                        // trusting `pubkey`/`id` for handler identity + correlation.
+                        // Redundant against the default RelayPool (it verifies inbound
+                        // signatures), but keeps caller identity independent of the
+                        // pool impl — a custom RelayPoolTrait may skip verification.
+                        if let Err(e) = event.verify() {
+                            tracing::warn!(
+                                target: LOG_TARGET,
+                                "Plaintext event signature verification failed: {e}"
+                            );
+                            continue;
+                        }
+                        let inbound = (*event).clone();
+                        (
+                            event.content.clone(),
+                            event.pubkey.to_hex(),
+                            event.id.to_hex(),
+                            false,
+                            event.tags.to_vec(),
+                            Some(inbound),
+                        )
+                    };
 
                 // Parse MCP message
                 let mcp_msg = match validation::validate_and_parse(&content) {
@@ -1857,6 +1945,7 @@ impl NostrServerTransport {
                                 &request_wrap_kinds,
                                 &tx,
                                 &open_stream,
+                                inbound_event,
                             )
                             .await;
                             continue;
@@ -1986,6 +2075,7 @@ impl NostrServerTransport {
                     client_pubkey: sender_pubkey,
                     event_id,
                     is_encrypted,
+                    event: inbound_event,
                 });
             }
         }
@@ -2016,6 +2106,7 @@ impl NostrServerTransport {
         request_wrap_kinds: &Arc<RwLock<HashMap<String, Option<u16>>>>,
         tx: &tokio::sync::mpsc::UnboundedSender<IncomingRequest>,
         open_stream: &ServerOpenStreamState,
+        inbound_event: Option<Event>,
     ) {
         // The outer progressToken keys the transfer (needed for accept + route).
         // String or number — defensive only: every known sender stringifies
@@ -2139,6 +2230,7 @@ impl NostrServerTransport {
                     client_pubkey: sender_pubkey.to_string(),
                     event_id: event_id.to_string(),
                     is_encrypted,
+                    event: inbound_event,
                 });
             }
             // Clean up locally, let the peer's own timeout fire.
@@ -2321,6 +2413,13 @@ impl NostrServerTransport {
             content_type: None,
             on_close: Some(on_close),
             on_abort: Some(on_abort),
+            // CEP-41 sender-side keepalive: reuse the reader idle/probe knobs (one
+            // idle/probe pair per stream, per spec). `None` when idle is disabled,
+            // keeping the writer keepalive in lockstep with the sweep gate
+            // (`open_stream_sweep_enabled` is false when `idle_timeout_ms == 0`).
+            idle_timeout: (state.policy.idle_timeout_ms != 0)
+                .then(|| Duration::from_millis(state.policy.idle_timeout_ms)),
+            probe_timeout: Duration::from_millis(state.policy.probe_timeout_ms),
         });
         let snapshot = RouteSnapshot {
             client_pubkey,
@@ -2409,6 +2508,29 @@ impl NostrServerTransport {
                     .await;
                 }
             }
+            Some(OpenStreamFrame::Pong { nonce }) => {
+                // CEP-41: a `pong` for a stream the server is *writing* (server→client)
+                // acknowledges our keepalive probe (completing the round trip the reader
+                // sweep opened). Only intercept when a writer owns the token;
+                // client→server streams fall through to the reader engine below.
+                if let Some(writer) = writer {
+                    writer.ack_probe(&nonce);
+                } else {
+                    Self::feed_open_stream_reader(
+                        state,
+                        relay_pool,
+                        encryption_mode,
+                        gift_wrap_mode,
+                        notification,
+                        sender_pubkey,
+                        event_id,
+                        is_encrypted,
+                        is_gift_wrap,
+                        outer_kind,
+                    )
+                    .await;
+                }
+            }
             Some(OpenStreamFrame::Start { .. }) => {
                 Self::feed_open_stream_reader(
                     state,
@@ -2604,6 +2726,7 @@ impl NostrServerTransport {
         relay_pool: &Arc<dyn RelayPoolTrait>,
         encryption_mode: EncryptionMode,
         gift_wrap_mode: GiftWrapMode,
+        sessions: &SessionStore,
     ) {
         let now = Instant::now();
         let mut actions: Vec<(String, String, KeepaliveAction)> = Vec::new();
@@ -2657,6 +2780,68 @@ impl NostrServerTransport {
                 KeepaliveAction::None => {}
             }
         }
+
+        // CEP-41: writer (server→client) keepalive — each producer stream MUST
+        // maintain an idle timeout (CEP-41 §Timeout and Keepalive). A client that
+        // vanishes without `abort` is probed here and aborted on a missing `pong`;
+        // the writer's `on_abort` hook then flushes any deferred final response.
+        // This is the mirror of the reader sweep above for sender-side streams.
+        // The client pubkey is captured up front: the slot may be removed by the
+        // time we act on an `Abort` (the writer's on_abort flush drops it), and we
+        // need it to release the dead client's session.
+        let writer_actions: Vec<(String, PublicKey, KeepaliveAction)> = state
+            .lock_slots()
+            .iter()
+            .map(|(event_id, slot)| {
+                (
+                    event_id.clone(),
+                    slot.snapshot.client_pubkey,
+                    slot.writer.tick(now),
+                )
+            })
+            .filter(|(.., action)| !matches!(action, KeepaliveAction::None))
+            .collect();
+        for (event_id, client_pubkey, action) in writer_actions {
+            let Some(writer) = state.writer_for(&event_id) else {
+                continue;
+            };
+            match action {
+                KeepaliveAction::SendPing(nonce) => {
+                    if let Err(error) = writer.send_probe(nonce).await {
+                        tracing::warn!(
+                            target: LOG_TARGET,
+                            error = %error,
+                            event_id = %event_id,
+                            "Failed to publish open-stream keepalive ping"
+                        );
+                    }
+                }
+                KeepaliveAction::Abort(reason) => {
+                    if let Err(error) = writer.abort(Some(reason)).await {
+                        tracing::warn!(
+                            target: LOG_TARGET,
+                            error = %error,
+                            event_id = %event_id,
+                            "Failed to abort open-stream writer on probe timeout"
+                        );
+                    }
+                    // CEP-41: a probe timeout means the client is gone — release
+                    // its session too ("release local state"), mirroring the TS
+                    // `handleProbeTimeout`. Done after the abort so the abort frame
+                    // is published first; eviction is independent of the
+                    // snapshot-backed deferred-response flush. Only the sweep
+                    // produces a writer `Abort`, so this path is probe-timeout only
+                    // (a tool-initiated `close`/`abort` does not evict the session).
+                    let pubkey_hex = client_pubkey.to_hex();
+                    if sessions.remove_session(&pubkey_hex).await {
+                        if let Some(cb) = sessions.eviction_callback() {
+                            cb(pubkey_hex);
+                        }
+                    }
+                }
+                KeepaliveAction::None => {}
+            }
+        }
     }
 
     async fn cleanup_sessions(
@@ -3481,6 +3666,8 @@ mod tests {
             content_type: None,
             on_close: None,
             on_abort: None,
+            idle_timeout: None,
+            probe_timeout: Duration::from_millis(20_000),
         })
     }
 
@@ -3624,4 +3811,117 @@ mod tests {
             "a disabled server must not expose writers (deferral never attempted)"
         );
     }
+
+    // ── CEP-41 writer keepalive sweep (server→client, silent-client abort) ──
+
+    /// A keepalive-armed writer whose publish closure is a no-op. The sweep
+    /// drives `tick`; a missing `pong` aborts it on the probe deadline.
+    fn keepalive_test_writer(token: &str, idle_ms: u64, probe_ms: u64) -> OpenStreamWriter {
+        let publish_frame: PublishFrame = Arc::new(|_frame: JsonRpcNotification| {
+            Box::pin(async move { Ok(EventId::all_zeros()) })
+        });
+        OpenStreamWriter::new(OpenStreamWriterOptions {
+            progress_token: token.to_string(),
+            publish_frame,
+            content_type: None,
+            on_close: None,
+            on_abort: None,
+            idle_timeout: Some(Duration::from_millis(idle_ms)),
+            probe_timeout: Duration::from_millis(probe_ms),
+        })
+    }
+
+    /// Regression: a client that silently disappears (sends no `pong`, no
+    /// `abort`) is detected by the server-writer keepalive sweep — the writer is
+    /// probed after the idle window and aborted once the probe times out, and the
+    /// dead client's session is evicted (CEP-41 "release local state", mirroring
+    /// the TS `handleProbeTimeout`). This is the rs-sdk port of the TS 0.13.8 fix;
+    /// without it the writer (and any upstream producer keyed on `is_active`)
+    /// leaks indefinitely. `tokio::time::sleep` guarantees *at least* the requested
+    /// duration, so the idle/probe margins are deterministic, not flaky.
+    #[tokio::test]
+    async fn sweep_aborts_writer_and_evicts_session_when_client_goes_silent() {
+        let config = NostrServerTransportConfig::default()
+            .with_open_stream(OpenStreamConfig::default().with_enabled(true));
+        let pool: Arc<dyn RelayPoolTrait> = Arc::new(MockRelayPool::new());
+        let mut transport = NostrServerTransport::with_relay_pool(config, pool)
+            .await
+            .expect("server transport");
+
+        let writer = keepalive_test_writer("tok-silent", 40, 60);
+        writer.start().await.expect("start the stream");
+        install_slot(&transport.open_stream, "evt-silent", writer.clone(), false);
+        assert!(writer.is_active(), "writer starts active");
+
+        // Install a session for the slot's client and arm the eviction callback.
+        let pubkey_hex = transport
+            .open_stream
+            .lock_slots()
+            .get("evt-silent")
+            .expect("slot")
+            .snapshot
+            .client_pubkey
+            .to_hex();
+        let evicted = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let evicted_cb = evicted.clone();
+        transport.sessions.set_eviction_callback(Arc::new(move |_| {
+            evicted_cb.store(true, std::sync::atomic::Ordering::SeqCst);
+        }));
+        transport
+            .sessions
+            .get_or_create_session(&pubkey_hex, false, &transport.event_routes)
+            .await;
+        assert!(
+            transport.sessions.get_session(&pubkey_hex).await.is_some(),
+            "session present before the sweep"
+        );
+
+        let relay_pool = Arc::clone(&transport.base.relay_pool);
+        let encryption_mode = transport.config.encryption_mode;
+        let gift_wrap_mode = transport.config.gift_wrap_mode;
+
+        // Past the idle window: the sweep probes (idle → SendPing). The writer is
+        // NOT aborted yet — it is waiting for a `pong` the silent client never sends.
+        tokio::time::sleep(Duration::from_millis(60)).await;
+        NostrServerTransport::sweep_open_stream_sessions(
+            &transport.open_stream,
+            &relay_pool,
+            encryption_mode,
+            gift_wrap_mode,
+            &transport.sessions,
+        )
+        .await;
+        assert!(
+            writer.is_active(),
+            "writer must remain active while a probe is in flight"
+        );
+        assert!(
+            transport.sessions.get_session(&pubkey_hex).await.is_some(),
+            "session must survive a mere probe (client not yet declared dead)"
+        );
+
+        // Past the probe timeout: the sweep aborts (probe deadline → Abort) and
+        // evicts the dead client's session.
+        tokio::time::sleep(Duration::from_millis(80)).await;
+        NostrServerTransport::sweep_open_stream_sessions(
+            &transport.open_stream,
+            &relay_pool,
+            encryption_mode,
+            gift_wrap_mode,
+            &transport.sessions,
+        )
+        .await;
+        assert!(
+            !writer.is_active(),
+            "writer must abort after a silent client misses the probe deadline"
+        );
+        assert!(
+            transport.sessions.get_session(&pubkey_hex).await.is_none(),
+            "dead client's session must be evicted on probe timeout"
+        );
+        assert!(
+            evicted.load(std::sync::atomic::Ordering::SeqCst),
+            "eviction callback must fire on probe-timeout session release"
+        );
+    }
 }
diff --git a/tests/e2e_happy_path.rs b/tests/e2e_happy_path.rs
index 529dea8..88f031d 100644
--- a/tests/e2e_happy_path.rs
+++ b/tests/e2e_happy_path.rs
@@ -12,7 +12,9 @@ use std::time::Duration;
 use contextvm_sdk::core::types::EncryptionMode;
 use contextvm_sdk::relay::mock::MockRelayPool;
 use contextvm_sdk::transport::client::{NostrClientTransport, NostrClientTransportConfig};
-use contextvm_sdk::transport::server::{NostrServerTransport, NostrServerTransportConfig};
+use contextvm_sdk::transport::server::{
+    ClientPubkey, InboundEvent, NostrServerTransport, NostrServerTransportConfig,
+};
 use contextvm_sdk::RelayPoolTrait;
 
 use rmcp::{
@@ -79,6 +81,33 @@ impl DemoServer {
             "Total echo calls: {n}"
         ))]))
     }
+
+    /// Self-check for caller-identity injection: returns the `ClientPubkey` the
+    /// rmcp worker places in request extensions, or `"none"` if absent.
+    #[tool(description = "Return the caller's Nostr pubkey from request extensions")]
+    async fn whoami(&self, ctx: RequestContext<RoleServer>) -> Result<CallToolResult, ErrorData> {
+        let pk = ctx
+            .extensions
+            .get::<ClientPubkey>()
+            .map(|c| c.0.clone())
+            .unwrap_or_else(|| "none".to_string());
+        Ok(CallToolResult::success(vec![Content::text(pk)]))
+    }
+
+    /// Self-check for inbound-event injection: returns `<id>|<sig>|<pubkey>`
+    /// from the `InboundEvent` the worker places in request extensions, or
+    /// `none` if no event was carried.
+    #[tool(description = "Return the inbound Nostr event id/sig/pubkey from extensions")]
+    async fn whoami_event(
+        &self,
+        ctx: RequestContext<RoleServer>,
+    ) -> Result<CallToolResult, ErrorData> {
+        let text = match ctx.extensions.get::<InboundEvent>() {
+            Some(ev) => format!("{}|{}|{}", ev.0.id.to_hex(), ev.0.sig, ev.0.pubkey.to_hex()),
+            None => "none".to_string(),
+        };
+        Ok(CallToolResult::success(vec![Content::text(text)]))
+    }
 }
 
 #[tool_handler]
@@ -159,8 +188,9 @@ fn call_params(name: &'static str, args: Option<serde_json::Value>) -> CallToolR
 async fn run_e2e_scenario(mode: EncryptionMode) {
     let (client_pool, server_pool) = MockRelayPool::create_pair();
 
-    // Extract server pubkey before as_pool() takes ownership
+    // Extract pubkeys before as_pool() takes ownership
     let server_pubkey_hex = server_pool.mock_public_key().to_hex();
+    let client_pubkey_hex = client_pool.mock_public_key().to_hex();
 
     let server_transport = NostrServerTransport::with_relay_pool(
         NostrServerTransportConfig::default().with_encryption_mode(mode),
@@ -207,7 +237,7 @@ async fn run_e2e_scenario(mode: EncryptionMode) {
     );
 
     let tools = client.list_all_tools().await.expect("list_all_tools");
-    assert_eq!(tools.len(), 3, "expected 3 tools");
+    assert_eq!(tools.len(), 5, "expected 5 tools");
     let tool_names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect();
     assert!(tool_names.contains(&"echo"), "missing echo tool");
     assert!(tool_names.contains(&"add"), "missing add tool");
@@ -215,6 +245,43 @@ async fn run_e2e_scenario(mode: EncryptionMode) {
         tool_names.contains(&"get_echo_count"),
         "missing get_echo_count tool"
     );
+    assert!(tool_names.contains(&"whoami"), "missing whoami tool");
+    assert!(
+        tool_names.contains(&"whoami_event"),
+        "missing whoami_event tool"
+    );
+
+    // ClientPubkey injection: the worker puts the caller's pubkey in extensions.
+    let whoami = client
+        .call_tool(call_params("whoami", None))
+        .await
+        .expect("whoami call");
+    assert_eq!(
+        first_text(&whoami),
+        client_pubkey_hex,
+        "whoami should return the caller's pubkey from request extensions"
+    );
+
+    // InboundEvent injection: the worker threads the full client-signed event
+    // (gift-wrap inner event when encrypted, outer event when plaintext), so
+    // its sig/id are reachable — the server cannot fabricate these on its own.
+    let ev = client
+        .call_tool(call_params("whoami_event", None))
+        .await
+        .expect("whoami_event call");
+    let ev_text = first_text(&ev);
+    assert_ne!(
+        ev_text, "none",
+        "InboundEvent must be present for real requests"
+    );
+    let parts: Vec<&str> = ev_text.split('|').collect();
+    assert_eq!(parts.len(), 3, "whoami_event format is id|sig|pubkey");
+    assert!(!parts[0].is_empty(), "event id must be non-empty");
+    assert!(!parts[1].is_empty(), "event sig must be non-empty");
+    assert_eq!(
+        parts[2], client_pubkey_hex,
+        "inbound event pubkey must match ClientPubkey"
+    );
 
     let echo1 = client
         .call_tool(call_params(
diff --git a/tests/transport_integration.rs b/tests/transport_integration.rs
index 4868c0e..25c52c7 100644
--- a/tests/transport_integration.rs
+++ b/tests/transport_integration.rs
@@ -389,6 +389,74 @@ async fn auth_allowlist_blocks_disallowed_pubkey() {
     );
 }
 
+// ── 4b. Plaintext signature verify blocks a forged (allowlisted) pubkey ─────
+//
+// A custom RelayPoolTrait that does NOT signature-verify inbound events (our
+// own MockRelayPool, or any third-party pool) lets a client put any pubkey in a
+// plaintext event. Without an in-app `verify()` that forgery bypasses the auth
+// allowlist — the gap TS fixed in ContextVM/sdk#64/#69. The plaintext arm must
+// verify the signature (and that id matches content) before trusting
+// `event.pubkey` for identity or correlation.
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn plaintext_forged_pubkey_is_rejected_by_signature_verify() {
+    let (_client_pool, server_pool) = MockRelayPool::create_pair();
+    let server_pubkey = server_pool.mock_public_key();
+    let server_pool = Arc::new(server_pool);
+
+    // The "victim" pubkey the attacker forges — it IS on the allowlist.
+    let victim_keys = Keys::generate();
+    // The attacker's real key, which actually signs the event.
+    let attacker_keys = Keys::generate();
+
+    let mut server = NostrServerTransport::with_relay_pool(
+        NostrServerTransportConfig::default()
+            .with_encryption_mode(EncryptionMode::Disabled)
+            .with_allowed_public_keys(vec![victim_keys.public_key().to_hex()]),
+        Arc::clone(&server_pool) as Arc<dyn RelayPoolTrait>,
+    )
+    .await
+    .expect("create server transport");
+
+    let mut server_rx = server
+        .take_message_receiver()
+        .expect("server message receiver");
+    server.start().await.expect("server start");
+    let_event_loops_start().await;
+
+    // Build a valid plaintext tools/list request addressed to the server,
+    // signed by the attacker, then forge the author pubkey to the victim's.
+    let request = JsonRpcMessage::Request(JsonRpcRequest {
+        jsonrpc: "2.0".to_string(),
+        id: serde_json::json!("forge-1"),
+        method: "tools/list".to_string(),
+        params: None,
+    });
+    let mut event = contextvm_sdk::core::serializers::mcp_to_nostr_event(
+        &request,
+        CTXVM_MESSAGES_KIND,
+        BaseTransport::create_recipient_tags(&server_pubkey),
+    )
+    .expect("serialize request")
+    .sign_with_keys(&attacker_keys)
+    .expect("sign event");
+    // The forgery: claim to be the allowlisted victim. The signature was made by
+    // the attacker, so an in-app `verify()` must reject this before the allowlist
+    // ever sees the (allowlisted) victim pubkey.
+    event.pubkey = victim_keys.public_key();
+    server_pool
+        .publish_event(&event)
+        .await
+        .expect("deliver forged event");
+
+    // The forged request must NOT reach the handler.
+    let result = tokio::time::timeout(Duration::from_millis(500), server_rx.recv()).await;
+    assert!(
+        result.is_err(),
+        "forged-pubkey plaintext event must be rejected by signature verification"
+    );
+}
+
 // ── 5. Encryption mode Required drops plaintext ─────────────────────────────
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -3660,6 +3728,7 @@ async fn server_close_stops_event_loop() {
 async fn oversized_request_roundtrip_client_to_server() {
     let (client_pool, server_pool) = MockRelayPool::create_pair();
     let server_pubkey = server_pool.mock_public_key();
+    let client_pubkey_hex = client_pool.mock_public_key().to_hex();
     let server_pool = Arc::new(server_pool);
 
     let oversized = OversizedTransferConfig::enabled()
@@ -3723,6 +3792,18 @@ async fn oversized_request_roundtrip_client_to_server() {
     assert_eq!(incoming.message.method(), Some("tools/call"));
     assert_eq!(incoming.message.id(), Some(&serde_json::json!("big-1")));
 
+    // CEP-22 reassembly surfaces the carrying event (parity with TS), so a
+    // handler can bind to / audit the chunked request rather than seeing `None`.
+    assert!(
+        incoming.event.is_some(),
+        "oversized request must carry its event"
+    );
+    assert_eq!(
+        incoming.event.as_ref().unwrap().pubkey.to_hex(),
+        client_pubkey_hex,
+        "carrying event pubkey must be the client's"
+    );
+
     // No second message should surface.
     let second = tokio::time::timeout(Duration::from_millis(100), server_rx.recv()).await;
     assert!(second.is_err(), "only one reassembled request expected");