diff --git a/CHANGELOG.md b/CHANGELOG.md
index 655d1708..a8c561dc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -45,6 +45,22 @@ policy in [docs/versioning.md](docs/versioning.md).
 
 ### Added
 
+- **A deadline for the coroutine receive** (#130).
+  `co_await stream.Receive(timeout)` on `AsyncEventStream` (and
+  `ReceiveMessage(socket, timeout)` for the raw awaitable) resolves with
+  `Error::Timeout` (code `TimeoutError`) when nothing arrives in time — the
+  completion-driven twin of the blocking `Receive(timeout)`, with the same
+  rule: a timeout is not terminal, the receive slot is released, and a
+  message the wire delivers after the deadline waits for the next await. A
+  `Detached` serve loop can now bound every await instead of parking its
+  frame forever on a peer that never sends. One layer down it is
+  `WebSocket::ReceiveAsync(timeout, callback)` — a refusing default (the
+  async family is opt-in), implemented by both transports (a timer on the
+  connection's executor for Beast; a watchdog for the executor-less
+  in-memory pair) and passed through by `JsonRpcStreamSocket`, with the
+  timeout/delivery race settled exactly once by a park generation. The
+  shared WebSocket contract suite grows four deadline cases every
+  implementation runs, in-repo and out.
 - **Servers validate intEnum membership** (#109). String enums already
   failed request validation outside the modeled value set; intEnum members
   were accepted silently. They now produce the same suite-exact
diff --git a/docs/production-guide.md b/docs/production-guide.md
index 2d24a221..775cee5b 100644
--- a/docs/production-guide.md
+++ b/docs/production-guide.md
@@ -461,6 +461,15 @@ code holding a raw session; a hand-rolled socket (a test fake, an adapter
 over another WebSocket library) owes its callers a wait that actually ends
 and an `Error::Timeout` when it does.
 
+The completion-driven half has the same deadline (#130):
+`co_await stream.Receive(std::chrono::seconds(2))` on an `AsyncEventStream`
+resolves with the same four outcomes and the same rule — a timeout releases
+the receive slot and leaves the session usable, and an event the wire
+delivers after the deadline waits for the next await. One layer down it is
+`WebSocket::ReceiveAsync(timeout, callback)`, a refusing default rather than
+a pure virtual (the async family is opt-in), with the shared contract suite
+holding every `SupportsAsync()` implementation to it.
+
 Not every `ClientConfig` knob reaches a streaming dial — the upgrade GET is
 not a unary request:
 
diff --git a/docs/server-guide.md b/docs/server-guide.md
index 2f4537dc..8655d174 100644
--- a/docs/server-guide.md
+++ b/docs/server-guide.md
@@ -140,6 +140,12 @@ while (true) {
 }
 ```
 
+The completion-driven seam has the same watchdog (#130): a `Detached` loop bounds its
+await with `co_await stream.Receive(std::chrono::seconds(1))` and gets the same
+`TimeoutError` on a session that keeps serving — the loop above works verbatim as a
+coroutine. `ReceiveMessage(socket, timeout)` bounds the raw pre-stream await the same
+way (the jsonRpc2 opening-envelope read).
+
 Whatever method the operation models, upgrades are always GET — a WebSocket upgrade is a
 GET on the wire, and the generated routes register accordingly: on the modeled URI for
 the binding protocols, on the shared `/` endpoint for jsonRpc2 (ADR-0023), whose opening
diff --git a/examples/bazel-consumer/async_acceptance_test.cc b/examples/bazel-consumer/async_acceptance_test.cc
index a0cb08aa..8c7a8e02 100644
--- a/examples/bazel-consumer/async_acceptance_test.cc
+++ b/examples/bazel-consumer/async_acceptance_test.cc
@@ -101,6 +101,50 @@ Detached Serve(smithy::server::SessionRegistry<Message>& registry, std::string i
   stream.Close();
 }
 
+// The #130 watchdog, out of tree and over a real socket: a hand-written
+// Detached loop — the deadline's stated audience — bounds its await
+// against a peer that says nothing unsolicited, treats the timeout as a
+// tick rather than an end, and the same coroutine still serves the
+// message that arrives after one.
+TEST(AsyncAcceptanceTest, AnAwaitedReceiveDeadlineTicksWithoutEndingTheSession) {
+  BeastServerTransport::Options options;
+  options.on_websocket = [](const HttpRequest&, WebSocket& socket) {
+    while (true) {
+      auto message = socket.Receive();
+      if (!message.ok() || !message->has_value()) return;
+      if (!socket.Send(Event("echo", "echo:" + (*message)->payload.ToString())).ok()) return;
+    }
+  };
+  BeastServerTransport server(options);
+  ASSERT_TRUE(server.Start(NotFound).ok());
+  auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()});
+  ASSERT_TRUE(dialed.ok()) << dialed.error().message();
+
+  std::promise<std::string> first_timeout_code;
+  std::promise<std::string> echoed;
+  [](std::shared_ptr<WebSocket> socket, std::promise<std::string>* timeout_code,
+     std::promise<std::string>* echoed) -> Detached {
+    AsyncEventStream<Message, Message> stream(std::move(socket), Identity, Identity);
+    // Quiet wire: the bounded await resolves with the timeout, terminally
+    // for nothing — the session and the loop both continue.
+    auto nothing = co_await stream.Receive(std::chrono::milliseconds(75));
+    timeout_code->set_value(nothing.ok() ? "ok" : nothing.error().code());
+    (void)co_await stream.Send(Event("chat", "after the deadline"));
+    auto echo = co_await stream.Receive(std::chrono::seconds(10));
+    echoed->set_value(echo.ok() && echo->has_value() ? (**echo).payload.ToString() : "<no echo>");
+    stream.Close();
+  }(*dialed, &first_timeout_code, &echoed);
+
+  auto code = first_timeout_code.get_future();
+  ASSERT_EQ(code.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  EXPECT_EQ(code.get(), "TimeoutError");
+
+  auto reply = echoed.get_future();
+  ASSERT_EQ(reply.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  EXPECT_EQ(reply.get(), "echo:after the deadline");
+  server.Stop();
+}
+
 TEST(AsyncAcceptanceTest, ThreeSessionsShareOneHandlerThreadAndAFanOutRegistry) {
   // Declared before the transport on purpose: sessions reference the
   // registry from their coroutines, so it must outlive them.
diff --git a/examples/bazel-consumer/model/todo.smithy b/examples/bazel-consumer/model/todo.smithy
index a10b6bdd..db4512ca 100644
--- a/examples/bazel-consumer/model/todo.smithy
+++ b/examples/bazel-consumer/model/todo.smithy
@@ -22,6 +22,10 @@ operation AddTask {
         @required
         @length(min: 1, max: 256)
         title: String
+
+        priority: Priority
+
+        effortHours: Float
     }
 
     output := {
@@ -55,6 +59,14 @@ operation GetTask {
     errors: [NoSuchTask]
 }
 
+/// int32 on the wire: out-of-range values fail the parse and unknown
+/// in-range values fail server validation (#109) — the members above give
+/// those checks out-of-tree, real-socket coverage.
+intEnum Priority {
+    LOW = 1
+    HIGH = 2
+}
+
 @error("client")
 @httpError(404)
 structure NoSuchTask {
diff --git a/examples/bazel-consumer/todo_beast_acceptance_test.cc b/examples/bazel-consumer/todo_beast_acceptance_test.cc
index 4151ee0b..64f20a95 100644
--- a/examples/bazel-consumer/todo_beast_acceptance_test.cc
+++ b/examples/bazel-consumer/todo_beast_acceptance_test.cc
@@ -107,6 +107,50 @@ TEST_F(TodoBeastAcceptanceTest, RoundTripsAndModeledErrorsWork) {
   ASSERT_NE(missing.error().detail<NoSuchTask>(), nullptr);
 }
 
+// The #109 numeric bounds, out of tree and over the real socket: hostile
+// wire values for the model's intEnum and float members are rejected by
+// the generated server before the handler runs, and the generated client
+// serializes valid ones — the same generator output consumers ship.
+TEST_F(TodoBeastAcceptanceTest, NumericBoundsAreEnforcedOverTheRealSocket) {
+  smithy::http::BeastHttpClient raw({.host = "127.0.0.1", .port = transport_->port()});
+  const auto post = [&raw](const std::string& body) {
+    smithy::http::HttpRequest request;
+    request.method = "POST";
+    request.target = "/tasks";
+    request.headers.Set("content-type", "application/json");
+    request.body = body;
+    return raw.Send(request);
+  };
+
+  // 2^32+2 would alias onto a valid Priority under a truncating cast; the
+  // parse rejects it instead.
+  auto aliased = post(R"({"title":"t","priority":4294967298})");
+  ASSERT_TRUE(aliased.ok()) << aliased.error().message();
+  EXPECT_EQ(aliased->status, 400);
+  EXPECT_EQ(aliased->headers.Get("x-error-type").value_or("<missing>"), "SerializationException");
+
+  // In range but outside the modeled set: membership validation, with the
+  // suite-exact message shape.
+  auto unknown = post(R"({"title":"t","priority":3})");
+  ASSERT_TRUE(unknown.ok()) << unknown.error().message();
+  EXPECT_EQ(unknown->status, 400);
+  EXPECT_EQ(unknown->headers.Get("x-error-type").value_or("<missing>"), "ValidationException");
+  EXPECT_NE(unknown->body.find("Member must satisfy enum value set: [1, 2]"), std::string::npos)
+      << unknown->body;
+
+  // A finite double beyond float range was UB to cast; it fails the parse.
+  auto overflow = post(R"({"title":"t","effortHours":1e300})");
+  ASSERT_TRUE(overflow.ok()) << overflow.error().message();
+  EXPECT_EQ(overflow->status, 400);
+  EXPECT_EQ(overflow->headers.Get("x-error-type").value_or("<missing>"), "SerializationException");
+
+  // Valid values ride the generated client end to end.
+  auto added = client_->AddTask(AddTaskInput{
+      .title = "well bounded", .priority = acme::todo::Priority::kHigh, .effortHours = 1.5F});
+  ASSERT_TRUE(added.ok()) << added.error().message();
+  EXPECT_EQ(added->title, "well bounded");
+}
+
 TEST_F(TodoBeastAcceptanceTest, LifecycleStopsAndRestartsAcrossTransportGenerations) {
   // The rolling-restart pattern from the production guide, composed entirely
   // from the consumer-visible API: serve, Stop() (bounded — the drain knob
diff --git a/examples/bazel-consumer/websocket_contract_consumer_test.cc b/examples/bazel-consumer/websocket_contract_consumer_test.cc
index 4587e78a..5a123f12 100644
--- a/examples/bazel-consumer/websocket_contract_consumer_test.cc
+++ b/examples/bazel-consumer/websocket_contract_consumer_test.cc
@@ -12,13 +12,16 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <string>
+#include <thread>
 #include <utility>
 
 #include "smithy/eventstream/frame.h"
@@ -37,7 +40,7 @@ using smithy::http::WebSocket;
 // wire but that the terminal transition is expressed with TerminalWaiters,
 // so this implementation inherits the ordering rule without its author
 // having to rediscover why the rule exists.
-class ConsumerSocket final : public WebSocket {
+class ConsumerSocket final : public WebSocket, public std::enable_shared_from_this<ConsumerSocket> {
  public:
   // Small on purpose: the contract suite wedges the wire by sending, and a
   // shallow queue gets there in a few messages.
@@ -74,17 +77,54 @@ class ConsumerSocket final : public WebSocket {
   // lock is released — the seam's documented shapes, nothing more.
   void ReceiveAsync(ReceiveCallback callback) override {
     Outcome<std::optional<Message>> immediate = std::optional<Message>();  // the clean end
+    // The same deliver-slot shape as the timed overload below: parking
+    // hands the completion to the session and empties this, so one park
+    // style covers both receives.
+    ReceiveCallback deliver = std::move(callback);
     {
       const std::lock_guard<std::mutex> lock(mutex_);
       if (!closed_ && !pending_receive_) {
-        pending_receive_ = std::move(callback);
-        return;  // EndSession completes it
+        pending_receive_ = std::exchange(deliver, nullptr);
+        ++receive_park_generation_;  // a stale deadline must not fire this park
+      } else if (!closed_) {
+        immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding");
       }
-      if (!closed_) {
+    }
+    if (!deliver) return;  // parked: EndSession completes it
+    deliver(std::move(immediate));
+  }
+
+  // The deadline overload (#130): the same park, bounded by a watchdog
+  // thread racing the terminal transition — the executor-less shape. The
+  // park generation settles the race exactly once under the lock, so a
+  // stale deadline can never time out a later receive; the watchdog holds
+  // the socket alive (shared_from_this) for at most its own deadline, and
+  // it is spawned outside the lock so a park never stalls the session
+  // behind pthread_create.
+  void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override {
+    Outcome<std::optional<Message>> immediate = std::optional<Message>();  // the clean end
+    // The completion this call still owes; parking hands it to the session
+    // and leaves this empty, so the slot itself says which happened.
+    ReceiveCallback deliver = std::move(callback);
+    std::uint64_t parked_generation = 0;  // 0 = not parked (the counter starts at 1)
+    {
+      const std::lock_guard<std::mutex> lock(mutex_);
+      if (!closed_ && !pending_receive_ && timeout > std::chrono::milliseconds::zero()) {
+        pending_receive_ = std::exchange(deliver, nullptr);
+        parked_generation = ++receive_park_generation_;
+      } else if (!closed_ && pending_receive_) {
         immediate = smithy::Error::Validation("consumer socket: a receive is already outstanding");
+      } else if (!closed_) {
+        // The non-positive poll — this peer never sends, so nothing is
+        // ever already in hand.
+        immediate = smithy::Error::Timeout("consumer socket: no message within the deadline");
       }
     }
-    callback(std::move(immediate));
+    if (!deliver) {
+      ArmDeadline(parked_generation, timeout);  // EndSession or the deadline completes it
+      return;
+    }
+    deliver(std::move(immediate));
   }
 
   void SendAsync(const Message& message, SendCallback callback) override {
@@ -126,10 +166,54 @@ class ConsumerSocket final : public WebSocket {
   }
 
  private:
+  // Arms the watchdog for the park `generation`. It sleeps on the shared
+  // condition variable, so a completed park (EndSession, or a fresh park's
+  // bumped generation) releases it early; at the deadline, a park still
+  // bearing its generation is timed out — the slot released exactly as a
+  // completion releases it. A spawn that fails must not leave the park
+  // unbounded: it is taken back (unless the session ended it first) and
+  // refused, keeping the callback exactly-once with nothing thrown at the
+  // caller. The saturation guards milliseconds::max()-style "practically
+  // forever" deadlines from overflowing into the past.
+  void ArmDeadline(std::uint64_t generation, std::chrono::milliseconds timeout) {
+    constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365);
+    const auto deadline = std::chrono::steady_clock::now() + std::min(timeout, kMaxWait);
+    const auto watchdog = [self = shared_from_this(), generation, deadline] {
+      ReceiveCallback expired;
+      {
+        std::unique_lock<std::mutex> lock(self->mutex_);
+        self->changed_.wait_until(lock, deadline, [&] {
+          return !self->pending_receive_ || self->receive_park_generation_ != generation;
+        });
+        if (!self->pending_receive_ || self->receive_park_generation_ != generation) {
+          return;  // the park this deadline bounded already completed
+        }
+        expired = std::exchange(self->pending_receive_, nullptr);
+        self->changed_.notify_all();
+      }
+      expired(smithy::Error::Timeout("consumer socket: no message within the deadline"));
+    };
+    ReceiveCallback refused;
+    try {
+      std::thread(watchdog).detach();
+      return;
+    } catch (...) {
+      const std::lock_guard<std::mutex> lock(mutex_);
+      if (pending_receive_ && receive_park_generation_ == generation) {
+        refused = std::exchange(pending_receive_, nullptr);
+        changed_.notify_all();
+      }
+    }
+    if (refused) {
+      refused(smithy::Error::Transport("consumer socket: cannot arm the receive deadline"));
+    }
+  }
+
   std::mutex mutex_;
   std::condition_variable changed_;
   ReceiveCallback pending_receive_;
   SendCallback pending_send_;
+  std::uint64_t receive_park_generation_ = 0;
   std::size_t queued_ = 0;
   bool closed_ = false;
 };
diff --git a/runtime/include/smithy/eventstream/async_event_stream.h b/runtime/include/smithy/eventstream/async_event_stream.h
index a00bb0b4..c5698540 100644
--- a/runtime/include/smithy/eventstream/async_event_stream.h
+++ b/runtime/include/smithy/eventstream/async_event_stream.h
@@ -3,6 +3,7 @@
 
 #include <atomic>
 #include <cassert>
+#include <chrono>
 #include <coroutine>
 #include <exception>
 #include <functional>
@@ -178,22 +179,29 @@ inline SendMessageAwaitable SendMessage(std::shared_ptr<http::WebSocket> socket,
 // One-outstanding-per-class passes through; single-shot like its twin.
 class [[nodiscard]] ReceiveMessageAwaitable {
  public:
-  explicit ReceiveMessageAwaitable(std::shared_ptr<http::WebSocket> socket)
-      : socket_(std::move(socket)) {}
+  explicit ReceiveMessageAwaitable(std::shared_ptr<http::WebSocket> socket,
+                                   std::optional<std::chrono::milliseconds> timeout = std::nullopt)
+      : socket_(std::move(socket)), timeout_(timeout) {}
   // NOLINTNEXTLINE(readability-convert-member-functions-to-static)
   bool await_ready() const noexcept { return false; }
   bool await_suspend(std::coroutine_handle<> coroutine) {
     // The second-arrival-resumes race, as in AsyncEventStream's awaitables.
-    socket_->ReceiveAsync([this, coroutine](Outcome<std::optional<Message>> message) {
+    auto completion = [this, coroutine](Outcome<std::optional<Message>> message) {
       received_ = std::move(message);
       if (arrived_.exchange(true)) coroutine.resume();
-    });
+    };
+    if (timeout_.has_value()) {
+      socket_->ReceiveAsync(*timeout_, std::move(completion));
+    } else {
+      socket_->ReceiveAsync(std::move(completion));
+    }
     return !arrived_.exchange(true);  // suspend iff the callback has not run
   }
   Outcome<std::optional<Message>> await_resume() noexcept { return std::move(received_); }
 
  private:
   std::shared_ptr<http::WebSocket> socket_;
+  std::optional<std::chrono::milliseconds> timeout_;
   std::atomic<bool> arrived_{false};
   // Outcome has no default constructor; the placeholder is overwritten
   // before any resume. NOLINT(readability-redundant-member-init)
@@ -204,6 +212,17 @@ inline ReceiveMessageAwaitable ReceiveMessage(std::shared_ptr<http::WebSocket> s
   return ReceiveMessageAwaitable(std::move(socket));
 }
 
+// The same await under a deadline (#130): a coroutine that must not park
+// forever on a peer that may never send. Resolves with the socket's timed
+// receive outcomes — the message, the clean close, the terminal error, or
+// Error::Timeout ("TimeoutError"), which is not terminal: the session and
+// its receive slot are exactly as a completed receive leaves them, so the
+// loop can await again, send, or close on its own schedule.
+inline ReceiveMessageAwaitable ReceiveMessage(std::shared_ptr<http::WebSocket> socket,
+                                              std::chrono::milliseconds timeout) {
+  return ReceiveMessageAwaitable(std::move(socket), timeout);
+}
+
 // The typed session's coroutine adapter (ADR-0019): EventStream's contract
 // over the completion-driven socket primitives, with `co_await` where the
 // blocking facade parks a thread. Owns its session (shared_ptr — the async
@@ -261,7 +280,9 @@ class AsyncEventStream {
   // is closed and the error is the awaited result.
   class [[nodiscard]] ReceiveAwaitable {
    public:
-    explicit ReceiveAwaitable(AsyncEventStream* stream) : stream_(stream) {}
+    explicit ReceiveAwaitable(AsyncEventStream* stream,
+                              std::optional<std::chrono::milliseconds> timeout = std::nullopt)
+        : stream_(stream), timeout_(timeout) {}
     bool await_ready() const noexcept { return false; }
     bool await_suspend(std::coroutine_handle<> coroutine) {
       // The completion may fire before this returns (immediate refusals,
@@ -272,13 +293,22 @@ class AsyncEventStream {
       // and only a truly asynchronous completion resumes from the callback.
       // The flag can live in the awaitable: the frame dies only after a
       // resume, and every resume is sequenced after both exchanges.
-      stream_->socket_->ReceiveAsync([this, coroutine](Outcome<std::optional<Message>> message) {
+      auto completion = [this, coroutine](Outcome<std::optional<Message>> message) {
         raw_ = std::move(message);
         if (arrived_.exchange(true)) coroutine.resume();
-      });
+      };
+      if (timeout_.has_value()) {
+        stream_->socket_->ReceiveAsync(*timeout_, std::move(completion));
+      } else {
+        stream_->socket_->ReceiveAsync(std::move(completion));
+      }
       return !arrived_.exchange(true);  // suspend iff the callback has not run
     }
     Outcome<std::optional<Rx>> await_resume() {
+      // A timeout (code "TimeoutError") arrives here as raw_'s error and
+      // passes through untouched: it is not terminal, nothing is closed,
+      // and the next co_await picks the stream up where it left off. Only
+      // a decoder rejection below ends the session.
       if (!raw_.ok()) return std::move(raw_).error();
       std::optional<Message>& message = *raw_;
       if (!message.has_value()) return std::optional<Rx>();
@@ -292,6 +322,7 @@ class AsyncEventStream {
 
    private:
     AsyncEventStream* stream_;
+    std::optional<std::chrono::milliseconds> timeout_;
     std::atomic<bool> arrived_{false};
     // Outcome has no default constructor; the placeholder is overwritten
     // before any resume. NOLINT(readability-redundant-member-init)
@@ -300,6 +331,23 @@ class AsyncEventStream {
 
   ReceiveAwaitable Receive() { return ReceiveAwaitable(this); }
 
+  // Receive under a deadline (#130) — the async twin of EventStream's
+  // timed Receive, for a loop that must not park forever on an event the
+  // peer may never send:
+  //
+  //   auto event = co_await stream.Receive(std::chrono::seconds(2));
+  //   if (!event.ok() && event.error().code() == "TimeoutError") { ... }
+  //
+  // Same outcomes plus Error::Timeout, and the same rule as the blocking
+  // overload: a timeout is not terminal — the session stays usable, the
+  // receive slot is released, and an event the wire delivers after the
+  // deadline waits in the session for the next await. A non-positive
+  // timeout polls. The deadline bounds THIS await only; the session's own
+  // idle timeout still governs a quiet wire.
+  ReceiveAwaitable Receive(std::chrono::milliseconds timeout) {
+    return ReceiveAwaitable(this, timeout);
+  }
+
   // Awaits one event onto the wire. Encoder failures surface without
   // suspending and leave the session untouched; wire failures are the
   // socket's (Error::Transport once the session ended). Uncallable when Tx
diff --git a/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h b/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h
index 5787d20b..ae1cb79c 100644
--- a/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h
+++ b/runtime/include/smithy/eventstream/jsonrpc_stream_socket.h
@@ -76,6 +76,11 @@ class JsonRpcStreamSocket final : public http::WebSocket {
   Outcome<Unit> Send(const Message& message) override;
   void Close() override;
   void ReceiveAsync(ReceiveCallback callback) override;
+  // The async deadline (#130), worn by the wrapper the way the blocking
+  // one is: a timeout is the inner Error::Timeout verbatim — no envelope
+  // classification runs on it, nothing is closed, and the stream picks up
+  // where it left off.
+  void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) override;
   void SendAsync(const Message& message, SendCallback callback) override;
   bool SupportsAsync() const override;
 
@@ -84,6 +89,11 @@ class JsonRpcStreamSocket final : public http::WebSocket {
   // answer a violation, close if the frame was fatal, hand back the rest.
   Outcome<std::optional<Message>> Police(Outcome<std::optional<Message>> raw);
 
+  // Both async receives: the classifying completion that Police-es one
+  // inbound outcome before handing it to `callback` — shared so the timed
+  // and untimed paths cannot drift.
+  ReceiveCallback PoliceAsync(ReceiveCallback callback);
+
   // Engaged only by the owning form; every call goes through inner_.
   std::shared_ptr<http::WebSocket> owner_;
   http::WebSocket* inner_;
diff --git a/runtime/include/smithy/http/websocket.h b/runtime/include/smithy/http/websocket.h
index fba27e10..5ac4a19b 100644
--- a/runtime/include/smithy/http/websocket.h
+++ b/runtime/include/smithy/http/websocket.h
@@ -198,6 +198,32 @@ class WebSocket {
         "check; the blocking Receive/Send still work)"));
   }
 
+  // The blocking deadline's completion-driven twin (#130): the callback
+  // fires exactly once with the same four outcomes as Receive(timeout) —
+  // the message, the peer's clean close (nullopt), the session's permanent
+  // error, or Error::Timeout ("TimeoutError") when the budget runs out
+  // with nothing to report. Exactly like the blocking overload, a timeout
+  // is not terminal: the session is untouched and usable, and the receive
+  // slot is released exactly as a completed receive releases it — the next
+  // receive (either API) may park again, and a message the wire delivers
+  // after the deadline waits in the session for it. A non-positive timeout
+  // polls: it completes without waiting — with what is already in hand
+  // (possibly on the completion context, like any ready result) or inline
+  // with the timeout. The one-outstanding-receive rule is unchanged.
+  //
+  // The timeout must race the completion and settle exactly once — the
+  // implementation owns that race the same way it owns the parked slot
+  // (there is nowhere honest to bound the wait from outside; the blocking
+  // overload's doc block explains why). The refusing default keeps every
+  // existing implementor compiling; the contract suite holds anything that
+  // reports SupportsAsync() to overriding this too.
+  virtual void ReceiveAsync(std::chrono::milliseconds timeout, ReceiveCallback callback) {
+    (void)timeout;
+    callback(Error::Validation(
+        "websocket: this implementation has no deadline-bounded async receive (the untimed "
+        "ReceiveAsync and the blocking Receive(timeout) may still work)"));
+  }
+
   virtual void SendAsync(const eventstream::Message& message, SendCallback callback) {
     (void)message;
     callback(Error::Validation(
diff --git a/runtime/src/eventstream/jsonrpc_stream_socket.cc b/runtime/src/eventstream/jsonrpc_stream_socket.cc
index f4c42850..f908feb9 100644
--- a/runtime/src/eventstream/jsonrpc_stream_socket.cc
+++ b/runtime/src/eventstream/jsonrpc_stream_socket.cc
@@ -125,34 +125,45 @@ Outcome<Unit> JsonRpcStreamSocket::Send(const Message& message) {
 void JsonRpcStreamSocket::Close() { inner_->Close(); }
 
 void JsonRpcStreamSocket::ReceiveAsync(ReceiveCallback callback) {
+  inner_->ReceiveAsync(PoliceAsync(std::move(callback)));
+}
+
+void JsonRpcStreamSocket::ReceiveAsync(std::chrono::milliseconds timeout,
+                                       ReceiveCallback callback) {
+  // A timeout arrives as !ok() and rides through the classifier untouched
+  // (no violation, no close) — the wrapper stays exactly as usable as the
+  // socket under it, mirroring the blocking deadline overload.
+  inner_->ReceiveAsync(timeout, PoliceAsync(std::move(callback)));
+}
+
+JsonRpcStreamSocket::ReceiveCallback JsonRpcStreamSocket::PoliceAsync(ReceiveCallback callback) {
   // State travels by value into the completion: the wrapper may be gone by
   // the time the transport completes. The owning form pins the inner
   // socket through `owner`; the borrowing form rides its documented
   // blocking-seam lifetime contract.
-  inner_->ReceiveAsync(
-      [id = id_, role = role_, owner = owner_, inner = inner_,
-       callback = std::move(callback)](Outcome<std::optional<Message>> raw) mutable {
-        Inbound inbound = ClassifyInbound(std::move(raw), id, role);
-        if (inbound.violation_text.has_value()) {
-          // The close AND the caller's resumption both ride the send's
-          // completion: the close cannot cancel the terminal write (the
-          // ADR-0021 lesson), and neither can the caller — a generated driver
-          // closes the session as it unwinds, which would escalate past the
-          // in-flight terminal — because it never runs until the frame is on
-          // the wire and the close is already requested.
-          auto terminal = std::make_shared<Message>();
-          terminal->payload = Blob::FromString(*std::move(inbound.violation_text));
-          inner->SendAsync(*terminal,
-                           [terminal, owner, inner, result = std::move(inbound.result),
-                            callback = std::move(callback)](const Outcome<Unit>& /*sent*/) mutable {
-                             inner->Close();
-                             callback(std::move(result));
-                           });
-          return;
-        }
-        if (inbound.close) inner->Close();
-        callback(std::move(inbound.result));
-      });
+  return [id = id_, role = role_, owner = owner_, inner = inner_,
+          callback = std::move(callback)](Outcome<std::optional<Message>> raw) mutable {
+    Inbound inbound = ClassifyInbound(std::move(raw), id, role);
+    if (inbound.violation_text.has_value()) {
+      // The close AND the caller's resumption both ride the send's
+      // completion: the close cannot cancel the terminal write (the
+      // ADR-0021 lesson), and neither can the caller — a generated driver
+      // closes the session as it unwinds, which would escalate past the
+      // in-flight terminal — because it never runs until the frame is on
+      // the wire and the close is already requested.
+      auto terminal = std::make_shared<Message>();
+      terminal->payload = Blob::FromString(*std::move(inbound.violation_text));
+      inner->SendAsync(*terminal,
+                       [terminal, owner, inner, result = std::move(inbound.result),
+                        callback = std::move(callback)](const Outcome<Unit>& /*sent*/) mutable {
+                         inner->Close();
+                         callback(std::move(result));
+                       });
+      return;
+    }
+    if (inbound.close) inner->Close();
+    callback(std::move(inbound.result));
+  };
 }
 
 void JsonRpcStreamSocket::SendAsync(const Message& message, SendCallback callback) {
diff --git a/runtime/src/http/beast_transport.cc b/runtime/src/http/beast_transport.cc
index c0f18a5d..f2ff9bc3 100644
--- a/runtime/src/http/beast_transport.cc
+++ b/runtime/src/http/beast_transport.cc
@@ -10,6 +10,7 @@
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/post.hpp>
 #include <boost/asio/ssl.hpp>
+#include <boost/asio/steady_timer.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/thread_pool.hpp>
 #include <boost/beast/core.hpp>
@@ -357,6 +358,24 @@ class WsSession final : public WebSocketSessionBase,
   // The completion-driven twin (ADR-0019): one outstanding receive-class
   // operation per session, completion on the connection's executor.
   void ReceiveAsync(WebSocket::ReceiveCallback callback) override {
+    ReceiveAsyncWithin(std::nullopt, std::move(callback));
+  }
+
+  // The deadline overload (#130): same immediate paths, and a parked
+  // receive races its deadline instead of waiting forever.
+  void ReceiveAsync(std::chrono::milliseconds timeout,
+                    WebSocket::ReceiveCallback callback) override {
+    ReceiveAsyncWithin(timeout, std::move(callback));
+  }
+
+  // Both async receives: `timeout` engaged bounds the park with a timer on
+  // the connection's executor, disengaged parks until Deliver or a
+  // terminal transition. The park generation settles the timer/delivery
+  // race exactly once under mutex_: every park bumps it, the timer fires
+  // only for the generation it was armed with, so a stale deadline can
+  // never time out a later receive (timed or not).
+  void ReceiveAsyncWithin(std::optional<std::chrono::milliseconds> timeout,
+                          WebSocket::ReceiveCallback callback) {
     Outcome<std::optional<eventstream::Message>> immediate = std::optional<eventstream::Message>();
     bool ready_message = false;
     {
@@ -376,9 +395,27 @@ class WsSession final : public WebSocketSessionBase,
         immediate = std::optional<eventstream::Message>();
       } else if (failed_) {
         immediate = Error::Transport("websocket: " + error_);
+      } else if (timeout.has_value() && *timeout <= std::chrono::milliseconds::zero()) {
+        // The blocking overload's poll shape: nothing already in hand is a
+        // timeout, inline like the other immediates below.
+        immediate = Error::Timeout("websocket: no message within the receive deadline");
       } else {
         pending_receive_ = std::move(callback);
-        return;  // Deliver / the terminal transitions complete it
+        ++receive_park_generation_;  // a stale deadline must not fire this park
+        if (timeout.has_value()) {
+          try {
+            ArmReceiveDeadlineLocked(*timeout);
+          } catch (...) {
+            // Could not schedule the deadline: an unbounded park would
+            // betray the caller's whole ask, so refuse instead.
+            WebSocket::ReceiveCallback refused = std::exchange(pending_receive_, nullptr);
+            wake_.notify_all();
+            lock.unlock();
+            refused(Error::Transport("websocket: cannot schedule the receive deadline"));
+            return;
+          }
+        }
+        return;  // Deliver / the deadline / the terminal transitions complete it
       }
     }
     if (!ready_message) {
@@ -403,6 +440,46 @@ class WsSession final : public WebSocketSessionBase,
     }
   }
 
+  // With mutex_ held: (re)arms the one receive-deadline timer for the park
+  // that just went in. Emplacing destroys a previous timer, cancelling its
+  // pending wait (the handler sees operation_aborted and returns) — and the
+  // generation check makes even an uncancelled stale wait harmless. The
+  // handler captures the session weakly: a deadline must never extend a
+  // session's life.
+  void ArmReceiveDeadlineLocked(std::chrono::milliseconds timeout) {
+    // Saturate far-future deadlines (milliseconds::max() as "practically
+    // forever") instead of overflowing the timer's now + duration into the
+    // past and firing a spurious instant timeout.
+    constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365);
+    receive_deadline_.emplace(ws_.get_executor());
+    receive_deadline_->expires_after(std::min(timeout, kMaxWait));
+    receive_deadline_->async_wait(
+        [weak = this->weak_from_this(),
+         generation = receive_park_generation_](const boost::system::error_code& ec) {
+          if (ec) return;  // cancelled: a newer park re-armed, or teardown
+          if (auto self = weak.lock()) self->OnReceiveDeadline(generation);
+        });
+  }
+
+  void OnReceiveDeadline(std::uint64_t generation) {
+    WebSocket::ReceiveCallback expired;
+    {
+      const std::lock_guard<std::mutex> lock(mutex_);
+      if (!pending_receive_ || receive_park_generation_ != generation) {
+        return;  // the park this deadline bounded already completed
+      }
+      expired = std::exchange(pending_receive_, nullptr);
+      // A blocking receiver may be waiting behind the parked slot; the
+      // freed slot is part of its wake condition.
+      wake_.notify_all();
+    }
+    // The timer fired on the connection's executor — the completion
+    // context — so contain a throwing application callback (ADR-0003).
+    InvokeCompletion("websocket receive", expired,
+                     Outcome<std::optional<eventstream::Message>>(
+                         Error::Timeout("websocket: no message within the receive deadline")));
+  }
+
   Outcome<Unit> Send(const eventstream::Message& message) override {
     auto frame = EncodeForWire(message);
     if (!frame.ok()) {
@@ -627,6 +704,8 @@ class WsSession final : public WebSocketSessionBase,
       const std::lock_guard<std::mutex> lock(mutex_);
       if (pending_receive_) {
         receive = std::exchange(pending_receive_, nullptr);
+        // The park is over, so its deadline has nothing left to bound.
+        receive_deadline_.reset();
         handoff.emplace(std::move(message));
       } else {
         received_.push_back(std::move(message));
@@ -649,6 +728,15 @@ class WsSession final : public WebSocketSessionBase,
   // can never fire twice and the slots' emptiness stays the busy signal.
   using AsyncWaiters = WebSocket::TerminalWaiters;
   AsyncWaiters TakeAsyncWaitersLocked() {
+    // Whatever deadline bounded the receive being taken is spent with it.
+    // Destroying the timer cancels its pending wait, so the handler returns
+    // on operation_aborted instead of waking the executor for a park that
+    // no longer exists; the generation guard already made such a wakeup a
+    // no-op, this just spares it. Safe from any thread for the same reason
+    // ArmReceiveDeadlineLocked's emplace is: every touch of the timer is
+    // under mutex_, and the armed handler captures the session weakly
+    // rather than the timer.
+    receive_deadline_.reset();
     return {std::exchange(pending_receive_, nullptr), std::exchange(pending_send_, nullptr)};
   }
 
@@ -873,6 +961,12 @@ class WsSession final : public WebSocketSessionBase,
   // transition takes and fires them exactly once (TakeAsyncWaitersLocked).
   WebSocket::ReceiveCallback pending_receive_;
   WebSocket::SendCallback pending_send_;
+  // Bumped on every pending_receive_ park; a receive deadline fires only
+  // for the generation it was armed with (the timed-receive race settles
+  // here). The timer is optional so it can be (re)constructed on the
+  // connection's executor per timed park.
+  std::uint64_t receive_park_generation_ = 0;
+  std::optional<asio::steady_timer> receive_deadline_;
   int blocked_receivers_ = 0;
   bool read_paused_ = false;
   bool peer_closed_ = false;  // clean close: Receive's nullopt
@@ -2237,6 +2331,10 @@ class DialedWebSocket final : public WebSocket {
   void ReceiveAsync(WebSocket::ReceiveCallback callback) override {
     session_->ReceiveAsync(std::move(callback));
   }
+  void ReceiveAsync(std::chrono::milliseconds timeout,
+                    WebSocket::ReceiveCallback callback) override {
+    session_->ReceiveAsync(timeout, std::move(callback));
+  }
   void SendAsync(const eventstream::Message& message, WebSocket::SendCallback callback) override {
     session_->SendAsync(message, std::move(callback));
   }
diff --git a/runtime/src/http/websocket_pair.cc b/runtime/src/http/websocket_pair.cc
index ae744ebb..ab72de58 100644
--- a/runtime/src/http/websocket_pair.cc
+++ b/runtime/src/http/websocket_pair.cc
@@ -1,12 +1,15 @@
 #include "smithy/http/websocket_pair.h"
 
+#include <algorithm>
 #include <array>
 #include <chrono>
 #include <condition_variable>
 #include <cstddef>
+#include <cstdint>
 #include <deque>
 #include <mutex>
 #include <optional>
+#include <thread>
 #include <utility>
 
 #include "smithy/eventstream/frame.h"
@@ -44,6 +47,10 @@ struct PairState {
   // two halves, same shape as the Beast session).
   std::array<WebSocket::ReceiveCallback, 2> pending_receive;
   std::array<std::optional<PendingSend>, 2> pending_send;
+  // Bumped on every pending_receive park; a receive-deadline watchdog fires
+  // only for the generation it was armed with, so a stale deadline can
+  // never time out a later receive (#130).
+  std::array<std::uint64_t, 2> receive_park_generation{};
   std::array<int, 2> blocked_receivers{};
   std::array<int, 2> blocked_senders{};
   bool closed = false;
@@ -132,30 +139,14 @@ class PairEnd final : public WebSocket {
   }
 
   void ReceiveAsync(WebSocket::ReceiveCallback callback) override {
-    WebSocket::SendCallback absorbed;
-    Outcome<std::optional<eventstream::Message>> immediate = std::optional<eventstream::Message>();
-    {
-      const std::lock_guard<std::mutex> lock(state_->mutex);
-      if (state_->pending_receive[send_index_] || state_->blocked_receivers[send_index_] > 0) {
-        callback(Error::Validation("websocket pair: a receive is already outstanding"));
-        return;
-      }
-      std::deque<eventstream::Message>& inbound = state_->queues[1 - send_index_];
-      if (!inbound.empty()) {
-        eventstream::Message message = std::move(inbound.front());
-        inbound.pop_front();
-        absorbed = AbsorbPeerPendingSendLocked();
-        state_->changed.notify_all();
-        immediate = std::optional<eventstream::Message>(std::move(message));
-      } else if (state_->closed) {
-        immediate = std::optional<eventstream::Message>();
-      } else {
-        state_->pending_receive[send_index_] = std::move(callback);
-        return;  // a send or the close completes it
-      }
-    }
-    if (absorbed) absorbed(Unit{});
-    callback(std::move(immediate));
+    ReceiveAsyncWithin(std::nullopt, std::move(callback));
+  }
+
+  // The deadline overload (#130): same immediate paths, and a parked
+  // receive races its deadline instead of waiting forever.
+  void ReceiveAsync(std::chrono::milliseconds timeout,
+                    WebSocket::ReceiveCallback callback) override {
+    ReceiveAsyncWithin(timeout, std::move(callback));
   }
 
   void SendAsync(const eventstream::Message& message, WebSocket::SendCallback callback) override {
@@ -195,6 +186,114 @@ class PairEnd final : public WebSocket {
   bool SupportsAsync() const override { return true; }
 
  private:
+  // Both async receives: `timeout` engaged bounds the park, disengaged
+  // parks until a send or the close completes it. The pair has no executor
+  // to run a timer on, so a timed park arms a detached watchdog thread —
+  // short-lived (it exits at the deadline or as soon as the park
+  // completes), and it owns nothing but the shared state it sleeps on. The
+  // park generation settles the watchdog/delivery race exactly once under
+  // the state mutex, so a stale deadline can never time out a later
+  // receive (timed or not).
+  void ReceiveAsyncWithin(std::optional<std::chrono::milliseconds> timeout,
+                          WebSocket::ReceiveCallback callback) {
+    WebSocket::SendCallback absorbed;
+    Outcome<std::optional<eventstream::Message>> immediate = std::optional<eventstream::Message>();
+    // The completion this call still owes, fired once the lock is released.
+    // Parking hands it to the session instead and leaves this empty (the
+    // std::exchange emptying is the same one the terminal paths rely on),
+    // so which of the two happened is read off the slot itself rather than
+    // off a flag a reader — or the analyzer — has to correlate with it.
+    WebSocket::ReceiveCallback deliver;
+    std::uint64_t parked_generation = 0;
+    {
+      const std::lock_guard<std::mutex> lock(state_->mutex);
+      if (state_->pending_receive[send_index_] || state_->blocked_receivers[send_index_] > 0) {
+        callback(Error::Validation("websocket pair: a receive is already outstanding"));
+        return;
+      }
+      deliver = std::move(callback);
+      std::deque<eventstream::Message>& inbound = state_->queues[1 - send_index_];
+      if (!inbound.empty()) {
+        eventstream::Message message = std::move(inbound.front());
+        inbound.pop_front();
+        absorbed = AbsorbPeerPendingSendLocked();
+        state_->changed.notify_all();
+        immediate = std::optional<eventstream::Message>(std::move(message));
+      } else if (state_->closed) {
+        immediate = std::optional<eventstream::Message>();
+      } else if (timeout.has_value() && *timeout <= std::chrono::milliseconds::zero()) {
+        // The blocking overload's poll shape: nothing already in hand is a
+        // timeout, completed inline like the other immediates.
+        immediate = Error::Timeout("websocket pair: no message within the receive deadline");
+      } else {
+        state_->pending_receive[send_index_] = std::exchange(deliver, nullptr);
+        parked_generation = ++state_->receive_park_generation[send_index_];
+      }
+    }
+    if (!deliver) {
+      // Parked: a send, the deadline, or the close completes it.
+      if (timeout.has_value()) ArmReceiveDeadline(parked_generation, *timeout);
+      return;
+    }
+    if (absorbed) absorbed(Unit{});
+    deliver(std::move(immediate));
+  }
+
+  // Spawns the watchdog for the park `generation` — outside the lock, so
+  // both ends' traffic never stalls behind pthread_create. It sleeps on
+  // the shared condition variable, so a completed park (delivery, close,
+  // or a fresh park's bumped generation) releases it early; at the
+  // deadline, a park still bearing its generation is timed out — the slot
+  // released exactly as a delivery releases it, the session untouched. A
+  // spawn that fails must not leave the park unbounded: the park is taken
+  // back (unless delivery already beat us to it) and refused, keeping the
+  // callback exactly-once with no exception escaping.
+  void ArmReceiveDeadline(std::uint64_t generation, std::chrono::milliseconds timeout) {
+    // Saturate far-future deadlines (milliseconds::max() as "practically
+    // forever") instead of overflowing now + timeout into the past.
+    constexpr std::chrono::milliseconds kMaxWait = std::chrono::hours(24 * 365);
+    const auto deadline = std::chrono::steady_clock::now() + std::min(timeout, kMaxWait);
+    const auto watchdog = [state = state_, end = send_index_, generation, deadline] {
+      WebSocket::ReceiveCallback expired;
+      {
+        std::unique_lock<std::mutex> lock(state->mutex);
+        state->changed.wait_until(lock, deadline, [&] {
+          return !state->pending_receive[end] || state->receive_park_generation[end] != generation;
+        });
+        if (!state->pending_receive[end] || state->receive_park_generation[end] != generation) {
+          return;  // the park this deadline bounded already completed
+        }
+        expired = std::exchange(state->pending_receive[end], nullptr);
+        // A blocking receiver may be waiting behind the parked slot; the
+        // freed slot is part of its wake condition.
+        state->changed.notify_all();
+      }
+      expired(Error::Timeout("websocket pair: no message within the receive deadline"));
+    };
+    // Under -fno-exceptions a failed thread spawn terminates (nothing can
+    // throw), which is the fail-fast posture; with exceptions on, contain
+    // it here so the caller never sees a throw beside a still-armed park.
+#if defined(__cpp_exceptions)
+    WebSocket::ReceiveCallback refused;
+    try {
+      std::thread(watchdog).detach();
+      return;
+    } catch (...) {
+      const std::lock_guard<std::mutex> lock(state_->mutex);
+      if (state_->pending_receive[send_index_] &&
+          state_->receive_park_generation[send_index_] == generation) {
+        refused = std::exchange(state_->pending_receive[send_index_], nullptr);
+        state_->changed.notify_all();
+      }
+    }
+    if (refused) {
+      refused(Error::Transport("websocket pair: cannot arm the receive deadline"));
+    }
+#else
+    std::thread(watchdog).detach();
+#endif
+  }
+
   // Both receive overloads: `timeout` engaged bounds the wait, disengaged
   // is the unbounded blocking call.
   Outcome<std::optional<eventstream::Message>> ReceiveWithin(
@@ -247,6 +346,9 @@ class PairEnd final : public WebSocket {
     WebSocket::ReceiveCallback& parked = state_->pending_receive[1 - send_index_];
     if (!parked) return nullptr;
     delivered = message;
+    // The emptied slot releases a deadline watchdog sleeping on it (#130);
+    // harmless for everyone else.
+    state_->changed.notify_all();
     return std::exchange(parked, nullptr);
   }
 
diff --git a/runtime/testing/include/smithy/testing/websocket_contract_test.h b/runtime/testing/include/smithy/testing/websocket_contract_test.h
index 04ba78c3..210f8596 100644
--- a/runtime/testing/include/smithy/testing/websocket_contract_test.h
+++ b/runtime/testing/include/smithy/testing/websocket_contract_test.h
@@ -335,6 +335,103 @@ TYPED_TEST_P(WebSocketContractTest, ASecondSendClassOperationRefusesWhileOneIsPa
   ender.join();
 }
 
+// The receive deadline's async half (#130): on a quiet wire the parked
+// receive completes with Error::Timeout — and, exactly like the blocking
+// overload, the timeout is not terminal: the slot is released (a second
+// timed receive parks and times out the same way) and the session still
+// answers a terminal transition afterwards.
+TYPED_TEST_P(WebSocketContractTest, ATimedReceiveOnAQuietWireTimesOutAndReleasesItsSlot) {
+  TypeParam driver;
+  std::shared_ptr<http::WebSocket> socket = driver.Socket();
+
+  for (int round = 0; round < 2; ++round) {
+    ContractMailbox<Outcome<std::optional<eventstream::Message>>> timed;
+    socket->ReceiveAsync(
+        std::chrono::milliseconds(50),
+        [&timed](Outcome<std::optional<eventstream::Message>> got) { timed.Post(std::move(got)); });
+    auto expired = timed.Wait();
+    ASSERT_FALSE(expired.ok()) << "round " << round;
+    EXPECT_EQ(expired.error().code(), "TimeoutError") << "round " << round;
+  }
+
+  // The session survived both deadlines: a parked receive still completes
+  // through the terminal transition, not as a leak.
+  ContractMailbox<Outcome<std::optional<eventstream::Message>>> parked;
+  socket->ReceiveAsync(
+      [&parked](Outcome<std::optional<eventstream::Message>> got) { parked.Post(std::move(got)); });
+  std::thread ender([&] { driver.EndSessionFromPeer(); });
+  ender.join();
+  auto terminal = parked.Wait();
+  if (!terminal.ok()) {
+    EXPECT_NE(terminal.error().code(), "TimeoutError");
+  }
+}
+
+// A non-positive deadline polls: nothing already in hand completes as a
+// timeout without waiting (the blocking overload's documented poll shape).
+TYPED_TEST_P(WebSocketContractTest, ANonPositiveTimedReceivePolls) {
+  TypeParam driver;
+  std::shared_ptr<http::WebSocket> socket = driver.Socket();
+  ContractMailbox<Outcome<std::optional<eventstream::Message>>> polled;
+
+  socket->ReceiveAsync(
+      std::chrono::milliseconds(0),
+      [&polled](Outcome<std::optional<eventstream::Message>> got) { polled.Post(std::move(got)); });
+  auto expired = polled.Wait();
+  ASSERT_FALSE(expired.ok());
+  EXPECT_EQ(expired.error().code(), "TimeoutError");
+
+  std::thread ender([&] { driver.EndSessionFromPeer(); });
+  ender.join();
+}
+
+// A terminal transition that arrives before the deadline owns the parked
+// timed receive: it completes with the session's terminal outcome, never
+// with the timeout — and the deadline that later fires into the emptied
+// slot must be a no-op, not a second completion (the mailbox asserts
+// exactly one arrival).
+TYPED_TEST_P(WebSocketContractTest, ATerminalTransitionBeatsAParkedTimedReceivesDeadline) {
+  TypeParam driver;
+  std::shared_ptr<http::WebSocket> socket = driver.Socket();
+  ContractMailbox<Outcome<std::optional<eventstream::Message>>> parked;
+
+  socket->ReceiveAsync(
+      std::chrono::seconds(30),
+      [&parked](Outcome<std::optional<eventstream::Message>> got) { parked.Post(std::move(got)); });
+  ASSERT_TRUE(parked.Empty()) << "the timed receive should park on a quiet wire";
+
+  std::thread ender([&] { driver.EndSessionFromPeer(); });
+  ender.join();
+  auto terminal = parked.Wait();
+  if (!terminal.ok()) {
+    EXPECT_NE(terminal.error().code(), "TimeoutError");
+  }
+}
+
+// The one-outstanding rule spans the deadline overload: a second receive
+// refuses while a timed one is parked, and the parked one still completes.
+TYPED_TEST_P(WebSocketContractTest, ASecondReceiveRefusesWhileATimedOneIsParked) {
+  TypeParam driver;
+  std::shared_ptr<http::WebSocket> socket = driver.Socket();
+  ContractMailbox<Outcome<std::optional<eventstream::Message>>> first;
+  ContractMailbox<Outcome<std::optional<eventstream::Message>>> second;
+
+  socket->ReceiveAsync(
+      std::chrono::seconds(30),
+      [&first](Outcome<std::optional<eventstream::Message>> got) { first.Post(std::move(got)); });
+  ASSERT_TRUE(first.Empty()) << "the timed receive should park on a quiet wire";
+
+  socket->ReceiveAsync(
+      [&second](Outcome<std::optional<eventstream::Message>> got) { second.Post(std::move(got)); });
+  auto refusal = second.Wait();
+  ASSERT_FALSE(refusal.ok());
+  EXPECT_EQ(refusal.error().kind(), ErrorKind::kValidation);
+
+  std::thread ender([&] { driver.EndSessionFromPeer(); });
+  ender.join();
+  (void)first.Wait();
+}
+
 // One outstanding receive-class operation per session, same shape.
 TYPED_TEST_P(WebSocketContractTest, ASecondReceiveClassOperationRefusesWhileOneIsParked) {
   TypeParam driver;
@@ -363,6 +460,10 @@ REGISTER_TYPED_TEST_SUITE_P(WebSocketContractTest,
                             ATerminalTransitionCompletesALoneParkedCoroutineSend,
                             ATerminalTransitionCompletesALoneParkedReceive,
                             ASecondSendClassOperationRefusesWhileOneIsParked,
+                            ATimedReceiveOnAQuietWireTimesOutAndReleasesItsSlot,
+                            ANonPositiveTimedReceivePolls,
+                            ATerminalTransitionBeatsAParkedTimedReceivesDeadline,
+                            ASecondReceiveRefusesWhileATimedOneIsParked,
                             ASecondReceiveClassOperationRefusesWhileOneIsParked);
 
 }  // namespace smithy::testing
diff --git a/runtime/tests/eventstream/async_event_stream_test.cc b/runtime/tests/eventstream/async_event_stream_test.cc
index 365fb830..fb2fb99d 100644
--- a/runtime/tests/eventstream/async_event_stream_test.cc
+++ b/runtime/tests/eventstream/async_event_stream_test.cc
@@ -73,11 +73,13 @@ template <typename T>
 class Mailbox {
  public:
   void Post(T value) {
-    {
-      const std::lock_guard<std::mutex> lock(mutex_);
-      ASSERT_FALSE(value_.has_value()) << "completion fired twice";
-      value_.emplace(std::move(value));
-    }
+    // Notify under the lock, not after it (ContractMailbox's rule): a
+    // poster on a foreign thread — a deadline watchdog, a peer completion —
+    // otherwise races the waiter's return and ~Mailbox runs while the
+    // poster is still inside notify_all.
+    const std::lock_guard<std::mutex> lock(mutex_);
+    ASSERT_FALSE(value_.has_value()) << "completion fired twice";
+    value_.emplace(std::move(value));
     ready_.notify_all();
   }
 
@@ -133,6 +135,84 @@ TEST(PairAsyncTest, AReadyMessageCompletesImmediately) {
   EXPECT_EQ((*outcome)->payload.ToString(), "1");
 }
 
+// The #130 deadline over the pair: what the transport-neutral contract
+// suite cannot pin because its driver has no "peer sends" hook.
+
+TEST(PairAsyncTest, ATimedReceiveDeliversAMessageThatBeatsTheDeadline) {
+  auto [a, b] = http::InMemoryWebSocketPair::Create();
+  Mailbox<Outcome<std::optional<Message>>> received;
+  a->ReceiveAsync(std::chrono::seconds(30), [&](Outcome<std::optional<Message>> message) {
+    received.Post(std::move(message));
+  });
+  EXPECT_TRUE(received.Empty());  // parked: nothing sent yet
+
+  ASSERT_TRUE(b->Send(RawPing(7)).ok());
+  auto outcome = received.Wait();
+  ASSERT_TRUE(outcome.ok() && outcome->has_value());
+  EXPECT_EQ((*outcome)->payload.ToString(), "7");
+}
+
+TEST(PairAsyncTest, ATimedReceiveTimesOutAndALaterMessageWaitsForTheNextReceive) {
+  auto [a, b] = http::InMemoryWebSocketPair::Create();
+  Mailbox<Outcome<std::optional<Message>>> timed;
+  a->ReceiveAsync(std::chrono::milliseconds(50),
+                  [&](Outcome<std::optional<Message>> message) { timed.Post(std::move(message)); });
+  auto expired = timed.Wait();
+  ASSERT_FALSE(expired.ok());
+  EXPECT_EQ(expired.error().code(), "TimeoutError");
+
+  // Nothing was lost to the deadline: the message the peer sends after it
+  // waits in the session for whoever receives next.
+  ASSERT_TRUE(b->Send(RawPing(9)).ok());
+  Mailbox<Outcome<std::optional<Message>>> next;
+  a->ReceiveAsync([&](Outcome<std::optional<Message>> message) { next.Post(std::move(message)); });
+  auto outcome = next.Wait();
+  ASSERT_TRUE(outcome.ok() && outcome->has_value());
+  EXPECT_EQ((*outcome)->payload.ToString(), "9");
+}
+
+TEST(PairAsyncTest, AStaleDeadlineNeverFiresALaterReceive) {
+  auto [a, b] = http::InMemoryWebSocketPair::Create();
+  // Park with a bounded deadline and complete it by delivery well inside
+  // it — a full second of margin, so a scheduler stall (TSan's slowdown on
+  // a loaded runner) cannot let the deadline win a race this test is not
+  // about.
+  Mailbox<Outcome<std::optional<Message>>> first;
+  a->ReceiveAsync(std::chrono::seconds(1),
+                  [&](Outcome<std::optional<Message>> message) { first.Post(std::move(message)); });
+  ASSERT_TRUE(b->Send(RawPing(1)).ok());
+  ASSERT_TRUE(first.Wait().ok());
+
+  // A fresh (untimed) park now occupies the slot the expired deadline was
+  // armed for. The generation guard makes the stale watchdog a no-op: well
+  // past the original deadline, the new park is still waiting.
+  Mailbox<Outcome<std::optional<Message>>> second;
+  a->ReceiveAsync(
+      [&](Outcome<std::optional<Message>> message) { second.Post(std::move(message)); });
+  std::this_thread::sleep_for(std::chrono::milliseconds(1300));
+  EXPECT_TRUE(second.Empty()) << "a stale deadline timed out a receive it never bounded";
+
+  ASSERT_TRUE(b->Send(RawPing(2)).ok());
+  auto outcome = second.Wait();
+  ASSERT_TRUE(outcome.ok() && outcome->has_value());
+  EXPECT_EQ((*outcome)->payload.ToString(), "2");
+}
+
+TEST(PairAsyncTest, AZeroTimeoutReceivePollsWhatIsAlreadyInHand) {
+  auto [a, b] = http::InMemoryWebSocketPair::Create();
+  ASSERT_TRUE(b->Send(RawPing(3)).ok());
+  Mailbox<Outcome<std::optional<Message>>> polled;
+  a->ReceiveAsync(std::chrono::milliseconds(0), [&](Outcome<std::optional<Message>> message) {
+    polled.Post(std::move(message));
+  });
+  // Inline: the pair has no executor, so the poll completed before the call
+  // returned — with the queued message, not a timeout.
+  ASSERT_FALSE(polled.Empty());
+  auto outcome = polled.Wait();
+  ASSERT_TRUE(outcome.ok() && outcome->has_value());
+  EXPECT_EQ((*outcome)->payload.ToString(), "3");
+}
+
 TEST(PairAsyncTest, ASecondOutstandingReceiveIsRefused) {
   auto [a, b] = http::InMemoryWebSocketPair::Create();
   Mailbox<Outcome<std::optional<Message>>> first;
@@ -358,6 +438,80 @@ TEST(AsyncEventStreamTest, ADetachedLoopEchoesAndEndsOnTheCleanClose) {
   EXPECT_TRUE(done.load());
 }
 
+// The #130 watchdog shape: a loop that bounds each await, treats the
+// timeout as "nothing yet" rather than an end, and keeps serving. The
+// deadline is not terminal — the same stream receives the ping that
+// arrives after a timeout and answers it.
+TEST(AsyncEventStreamTest, AwaitedReceiveTimesOutAndTheSessionStaysUsable) {
+  auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create();
+  std::atomic<int> timeouts{0};
+  std::atomic<bool> done{false};
+
+  [](std::shared_ptr<http::WebSocket> socket, std::atomic<int>* timeouts,
+     std::atomic<bool>* done) -> Detached {
+    AsyncServer stream(std::move(socket), EncodePong, DecodePing);
+    while (true) {
+      auto ping = co_await stream.Receive(std::chrono::milliseconds(50));
+      if (!ping.ok()) {
+        if (ping.error().code() == "TimeoutError") {
+          ++*timeouts;  // not terminal: the watchdog tick, then wait again
+          continue;
+        }
+        break;
+      }
+      if (!ping->has_value()) break;
+      auto sent = co_await stream.Send(Pong{"pong-" + std::to_string((*ping)->number)});
+      if (!sent.ok()) break;
+    }
+    *done = true;
+  }(server_socket, &timeouts, &done);
+
+  // Let at least one deadline expire before the first ping.
+  for (int i = 0; i < 100 && timeouts.load() == 0; ++i) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+  EXPECT_GT(timeouts.load(), 0) << "the awaited deadline never fired";
+  EXPECT_FALSE(done.load()) << "a timeout must not end the loop";
+
+  EventStream<Ping, Pong> client(client_socket, EncodePing, DecodePong);
+  ASSERT_TRUE(client.Send(Ping{42}).ok());
+  auto pong = client.Receive();
+  ASSERT_TRUE(pong.ok() && pong->has_value());
+  EXPECT_EQ((*pong)->text, "pong-42");
+
+  client.Close();
+  for (int i = 0; i < 100 && !done.load(); ++i) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+  EXPECT_TRUE(done.load());
+}
+
+// The raw awaitable's deadline (#130): ReceiveMessage with a timeout — the
+// ADR-0023 launch-body shape, where a client that never sends its opening
+// envelope must not park the serve coroutine forever.
+TEST(AsyncEventStreamTest, ReceiveMessageHonorsItsDeadline) {
+  auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create();
+  Mailbox<Outcome<std::optional<Message>>> first;
+  Mailbox<Outcome<std::optional<Message>>> second;
+
+  [](std::shared_ptr<http::WebSocket> socket, Mailbox<Outcome<std::optional<Message>>>* first,
+     Mailbox<Outcome<std::optional<Message>>>* second) -> Detached {
+    first->Post(co_await ReceiveMessage(socket, std::chrono::milliseconds(50)));
+    // The timeout released the slot: the same socket awaits again and gets
+    // the envelope that arrives late.
+    second->Post(co_await ReceiveMessage(socket, std::chrono::seconds(30)));
+  }(server_socket, &first, &second);
+
+  auto expired = first.Wait();
+  ASSERT_FALSE(expired.ok());
+  EXPECT_EQ(expired.error().code(), "TimeoutError");
+
+  ASSERT_TRUE(client_socket->Send(RawPing(5)).ok());
+  auto arrived = second.Wait();
+  ASSERT_TRUE(arrived.ok() && arrived->has_value());
+  EXPECT_EQ((*arrived)->payload.ToString(), "5");
+}
+
 TEST(AsyncEventStreamTest, AwaitedSendBackpressuresWithoutAThread) {
   auto [client_socket, server_socket] = http::InMemoryWebSocketPair::Create();
   std::atomic<int> sent_count{0};
diff --git a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc
index 5cb86152..35fcd743 100644
--- a/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc
+++ b/runtime/tests/eventstream/jsonrpc_stream_socket_test.cc
@@ -13,6 +13,7 @@
 #include <gtest/gtest.h>
 
 #include <chrono>
+#include <future>
 #include <memory>
 #include <optional>
 #include <string>
@@ -320,6 +321,39 @@ TEST(JsonRpcStreamSocketTest, ADeadlineDelegatesAndATimeoutIsNotAViolation) {
   EXPECT_EQ(**received, event);
 }
 
+TEST(JsonRpcStreamSocketTest, TheAsyncDeadlineDelegatesAndATimeoutIsNotAViolation) {
+  // The #130 twin of the blocking case above: the timeout rides through
+  // the classifier untouched, and the envelope that arrives after it is
+  // policed for the next receive exactly as if no deadline had passed.
+  auto [left, right] = http::InMemoryWebSocketPair::Create();
+  auto client = Wrap(left, JsonRpcStreamSocket::Role::kClient);
+  auto server = Wrap(right, JsonRpcStreamSocket::Role::kServer);
+
+  std::promise<Outcome<std::optional<Message>>> timed;
+  server->ReceiveAsync(std::chrono::milliseconds(50), [&timed](Outcome<std::optional<Message>> m) {
+    timed.set_value(std::move(m));
+  });
+  auto timed_future = timed.get_future();
+  ASSERT_EQ(timed_future.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  auto nothing = timed_future.get();
+  ASSERT_FALSE(nothing.ok());
+  EXPECT_EQ(nothing.error().code(), "TimeoutError");
+
+  const Message event =
+      MakeEventMessage("message", "application/json", Blob::FromString(R"({"text":"late"})"));
+  ASSERT_TRUE(client->Send(event).ok());
+  std::promise<Outcome<std::optional<Message>>> next;
+  server->ReceiveAsync(std::chrono::seconds(30), [&next](Outcome<std::optional<Message>> m) {
+    next.set_value(std::move(m));
+  });
+  auto next_future = next.get_future();
+  ASSERT_EQ(next_future.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  auto received = next_future.get();
+  ASSERT_TRUE(received.ok()) << received.error().message();
+  ASSERT_TRUE(received->has_value());
+  EXPECT_EQ(**received, event);
+}
+
 TEST(JsonRpcStreamSocketTest, CloseAndThePeersCleanCloseDelegate) {
   auto [left, right] = http::InMemoryWebSocketPair::Create();
   auto client = Wrap(left, JsonRpcStreamSocket::Role::kClient);
diff --git a/runtime/tests/http/beast_websocket_test.cc b/runtime/tests/http/beast_websocket_test.cc
index 31a70df0..19ffa5b1 100644
--- a/runtime/tests/http/beast_websocket_test.cc
+++ b/runtime/tests/http/beast_websocket_test.cc
@@ -172,6 +172,48 @@ TEST(BeastWebSocketTest, ATimedOutReceiveReleasesTheOneOutstandingSlotOnTheWire)
   server.Stop();
 }
 
+TEST(BeastWebSocketTest, AnAsyncReceiveDeadlineExpiresAndDeliveryStillBeatsALaterOne) {
+  // The #130 async twin over the real wire: the deadline timer lives on
+  // the connection's executor, so the timer/delivery race the pair pins in
+  // memory gets its own proof against Beast's io machinery.
+  BeastServerTransport server(EchoOptions());
+  ASSERT_TRUE(server.Start(NotFoundHandler()).ok());
+
+  auto dialed = BeastWebSocketClient::Dial({.host = "127.0.0.1", .port = server.port()});
+  ASSERT_TRUE(dialed.ok()) << dialed.error().message();
+  const std::shared_ptr<WebSocket>& socket = *dialed;
+
+  // Quiet wire: the parked timed receive completes with the timeout, on
+  // the executor, and the session is untouched.
+  std::promise<Outcome<std::optional<Message>>> timed;
+  socket->ReceiveAsync(std::chrono::milliseconds(75), [&timed](Outcome<std::optional<Message>> m) {
+    timed.set_value(std::move(m));
+  });
+  auto timed_future = timed.get_future();
+  ASSERT_EQ(timed_future.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  auto expired = timed_future.get();
+  ASSERT_FALSE(expired.ok());
+  EXPECT_EQ(expired.error().code(), "TimeoutError");
+
+  // Same session, fresh deadline: the echo arrives well inside it, the
+  // delivery wins the race, and the stale first timer never fires into
+  // this park (set_value would abort on a second completion).
+  std::promise<Outcome<std::optional<Message>>> delivered;
+  socket->ReceiveAsync(std::chrono::seconds(30), [&delivered](Outcome<std::optional<Message>> m) {
+    delivered.set_value(std::move(m));
+  });
+  ASSERT_TRUE(socket->Send(Text("chat", "beats the deadline")).ok());
+  auto delivered_future = delivered.get_future();
+  ASSERT_EQ(delivered_future.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  auto echo = delivered_future.get();
+  ASSERT_TRUE(echo.ok()) << echo.error().message();
+  ASSERT_TRUE(echo->has_value());
+  EXPECT_EQ((**echo).payload.ToString(), "echo:beats the deadline");
+
+  socket->Close();
+  server.Stop();
+}
+
 TEST(BeastWebSocketTest, AThrowingAsyncReceiveCallbackDoesNotKillTheSession) {
   // ADR-0003/#109: an application completion callback that throws runs on the
   // connection's io thread. It must be contained there — never unwind