diff --git a/CMakeLists.txt b/CMakeLists.txt
index 73212ee..1e8ea78 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -89,6 +89,7 @@ if(GTest_FOUND)
         tests/unit/ReplaceTest.cpp
         tests/unit/ContainsKeyTest.cpp
         tests/unit/HashAwareRoutingTest.cpp
+        tests/unit/ErrorHandlingTest.cpp
     )
 
     target_link_libraries(unit_tests
@@ -155,6 +156,10 @@ if(GTest_FOUND)
         tests/integration/ContainsKeyIntegrationTest.cpp
     )
 
+    add_executable(errorhandling_integration_tests
+        tests/integration/ErrorHandlingIntegrationTest.cpp
+    )
+
     # Topology integration tests (multi-server cluster)
     add_executable(topology_integration_tests
         tests/integration/TopologyChangeTest.cpp
@@ -235,6 +240,13 @@ if(GTest_FOUND)
             # Note: NOT gtest_main - we provide our own main()
     )
 
+    target_link_libraries(errorhandling_integration_tests
+        PRIVATE
+            hotrod-client
+            GTest::gtest
+            # Note: NOT gtest_main - we provide our own main()
+    )
+
     target_link_libraries(topology_integration_tests
         PRIVATE
             hotrod-client
@@ -283,6 +295,7 @@ if(GTest_FOUND)
     add_test(NAME PutIfAbsentIntegrationTests COMMAND putifabsent_integration_tests)
     add_test(NAME ReplaceIntegrationTests COMMAND replace_integration_tests)
     add_test(NAME ContainsKeyIntegrationTests COMMAND containskey_integration_tests)
+    add_test(NAME ErrorHandlingIntegrationTests COMMAND errorhandling_integration_tests)
     add_test(NAME TopologyIntegrationTests COMMAND topology_integration_tests)
     add_test(NAME ConcurrentClientsTests COMMAND concurrent_clients_tests)
     add_test(NAME HashAwareRoutingIntegrationTests COMMAND hash_aware_routing_integration_tests)
@@ -329,6 +342,10 @@ if(GTest_FOUND)
         WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
         LABELS "integration;docker"
     )
+    set_tests_properties(ErrorHandlingIntegrationTests PROPERTIES
+        WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
+        LABELS "integration;docker"
+    )
     set_tests_properties(TopologyIntegrationTests PROPERTIES
         WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
         LABELS "integration;docker;topology;multi-server"
diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md
index 70f4052..f3dc915 100644
--- a/docs/DECISIONS.md
+++ b/docs/DECISIONS.md
@@ -173,3 +173,59 @@ hook (rejected — a warning is just convention with extra steps); leaving
 history in STATUS.md (rejected — it's the source of the drift). Existing
 historical narrative already in STATUS.md is left in place (not migrated); the
 split is forward-looking.
+
+## 2026-09-25 — Step 11a shipped: two-axis error model; 0x86 transient (diverges from Java)
+
+**What:** Step 11a (error surfacing) is implemented and tested (11 unit + 3
+integration; full suite green: 187 unit, 78 integration across 15 suites).
+`HotRodClientException` (in `include/hotrod/HotRodClientException.h`) is a
+pure-data exception extending `std::runtime_error` with `FailurePhase phase`
+{BeforeSend, AfterSend, ServerError}, `std::optional<uint8_t> serverStatus`,
+`std::vector<ServerAddress> triedNodes`, and `bool ownersExhausted`. Server
+ERROR responses (opcode 0x50) are now parsed and surfaced; `RemoteCache`'s bare
+`throw std::runtime_error(...)` sites became typed throws
+(`FailurePhase::ServerError` + status). Status constants `REQUEST_PARSING_ERROR`
+(0x84), `NODE_SUSPECTED` (0x87), `ILLEGAL_LIFECYCLE_STATE` (0x88) were added to
+`HeaderCodec.h`.
+
+**Decisions (refine the 2026-09-21 design):**
+1. **Two orthogonal axes, no stored `retriable` bool.** Instead of baking a
+   retriable flag into the exception, classification is derived by free helpers:
+   `isTransient(e)` = *futility* ("is a retry worth it?" — the library's call)
+   and `outcomeUncertain(e)` = *ambiguity/safety* ("might the op have applied?"
+   — the user's call). `isTransient`: BeforeSend & AfterSend → true;
+   ServerError → `isTransientStatus(status)` (0x86/0x87/0x88 transient,
+   0x81–0x85 permanent). `outcomeUncertain`: AfterSend, or ServerError with
+   0x86. This replaces the design's earlier `retriable` field.
+2. **`COMMAND_TIMEOUT` (0x86) is transient *and* outcome-uncertain — intentionally
+   diverges from Java** (Java does not treat it as retriable). Rationale: with
+   the two-axis model the ambiguity is carried separately by
+   `outcomeUncertain()`, so marking 0x86 transient just says "a retry may
+   succeed elsewhere" without hiding that the op may have already applied — the
+   user still decides via `outcomeUncertain()`.
+3. **`proxyToNonOwner` client config, default `true`** (matches Java's non-owner
+   fallback via `OperationDispatcher.addressForObject` → `balancer.nextServer`).
+   Deferred to 11b (it governs retry routing, not surfacing).
+
+**Why it matters / gotcha found:** the old read loop treated ERROR 0x50 as an
+"opcode mismatch" **and left the length-prefixed error message unread**,
+desyncing the socket for every subsequent response on that connection. The fix
+drains the ERROR body in `readLoop` *before* the pending-request lookup (so even
+an orphan ERROR keeps the stream in sync). Proven by the
+`ConnectionUsableAfterServerError` integration test: trigger an ERROR on a
+missing cache, then PUT/GET round-trip on a real cache over the *same*
+connection succeeds. (Also learned along the way: the server's default cache
+isn't configured, so a missing/undefined cache returns status 0x84
+`CacheNotFoundException` — permanent under the classification above.)
+
+**Java cross-check (Explore):** non-owner fallback confirmed; Java default
+`max_retries` = 3 (`ConfigurationProperties.DEFAULT_MAX_RETRIES`); Java's
+retriable set is `RemoteNodeSuspectException` (0x87),
+`RemoteIllegalLifecycleStateException` (0x88), and `TransportException`
+(no status) — we additionally treat 0x86 as transient per decision 2, and make
+retry user-decided rather than automatic (2026-09-21 entry).
+
+**Alternatives:** store a `retriable` bool on the exception (rejected — conflates
+the two axes and forces the library to pre-judge user idempotency); follow Java
+and treat 0x86 as non-retriable (rejected — the ambiguity is already carried by
+`outcomeUncertain()`, so the extra futility signal is safe and useful).
diff --git a/docs/ERROR_HANDLING_DESIGN.md b/docs/ERROR_HANDLING_DESIGN.md
index b8c4428..d1b11df 100644
--- a/docs/ERROR_HANDLING_DESIGN.md
+++ b/docs/ERROR_HANDLING_DESIGN.md
@@ -5,7 +5,9 @@
 > lands; record *why* decisions were made in [`DECISIONS.md`](DECISIONS.md) and
 > keep the one-line pointer in [`STATUS.md`](STATUS.md) → *Next steps* current.
 >
-> **Created:** 2026-09-21 · **Status:** design agreed, implementation not started
+> **Created:** 2026-09-21 · **Refined:** 2026-09-25 (D4 classification, D5
+> proxy-to-non-owner, Java cross-check) · **Status:** design agreed,
+> implementation not started
 
 ---
 
@@ -74,6 +76,45 @@ lifetime-hazard class (an error object that captures client internals and can
 outlive them); 1.b keeps the error a trivially copyable value and pays only a
 mechanical, defaultable extra parameter on each op.
 
+### D4 — Classify *futility* (transient vs permanent), not *safety* (2026-09-25)
+Two orthogonal questions decide a retry: **(1) is it safe?** (won't corrupt the
+logical answer) and **(2) is it worth it?** (could the failure mode plausibly
+clear). Only (1) can be left to the user — it needs op-idempotency knowledge the
+library lacks (D2). (2) needs the *error taxonomy*, which only the library has;
+forcing users to reconstruct it from raw status codes recreates the
+string-matching pain we're removing. So the library **classifies futility**, the
+user **decides safety**, and a retry is warranted only when *both* hold.
+
+To keep this honest we avoid the word "retriable" (it sounds like a safety
+verdict). The library exposes raw facts (`phase`, `serverStatus`) and an
+**advisory** classifier `isTransient()` (§3.3). Classification:
+- **`phase == BeforeSend`** → transient (nothing sent; safe *and* not futile).
+- **`phase == AfterSend`** → transient-but-**ambiguous** (futility says "not
+  pointless", but this is the case where the user's idempotency judgment gates).
+- **`phase == ServerError`** → classify by status code (§3.3 table). Node-level
+  codes (`0x87 NODE_SUSPECTED`, `0x88 ILLEGAL_LIFECYCLE_STATE`) and `0x86
+  COMMAND_TIMEOUT` are transient (`0x86` also flagged `outcomeUncertain`);
+  request-level codes (`0x81–0x85`) are permanent.
+
+### D5 — `proxyToNonOwner`, defaulted `true` (matches Java) (2026-09-25)
+When a key's owner(s) can't be reached, do we auto-fall-back to a **non-owner**
+(which proxies to the owner), or stop at "owners exhausted" and let the user opt
+in? Made configurable via a flag **defaulted to `true`**, matching Java.
+
+Java-confirmed (2026-09-25, `OperationDispatcher.addressForObject` /
+`RoundRobinBalancingStrategy`): once an owner is in the op's `failedServers`,
+routing falls through to `balancer.nextServer(...)` over the *whole* cluster; the
+chosen non-owner proxies to the real owner. Default `max_retries = 3`
+(`ConfigurationProperties.DEFAULT_MAX_RETRIES`).
+
+**Scope caveat:** in *our* model the flag governs only the **BeforeSend**
+connection-fallback inside `selectServerForKey`. Java also auto-re-dispatches
+`AfterSend` transport failures (accepting double-apply risk); we deliberately do
+**not** (D2). So `proxyToNonOwner=true` means "Java-like *routing* fallback", not
+"become fully Java-automatic". `false` → after the owner loop fails, throw a
+transient exception with `ownersExhausted = true` so the user drives the proxy
+step.
+
 ---
 
 ## 3. The design (target)
@@ -87,14 +128,22 @@ enum class FailurePhase { BeforeSend, AfterSend, ServerError };
 
 class HotRodClientException : public std::runtime_error {
 public:
-    bool                       retriable;      // is a retry permitted at all?
-    FailurePhase               phase;          // WHEN it failed — see below
-    std::vector<ServerAddress> triedNodes;     // nodes already attempted
-    std::optional<uint8_t>     serverStatus;   // Hot Rod status / ERROR code, if any
+    FailurePhase               phase;          // WHEN it failed — see below (raw fact)
+    std::vector<ServerAddress> triedNodes;     // nodes attempted (UNION across retries)
+    std::optional<uint8_t>     serverStatus;   // Hot Rod status / ERROR code, if any (raw)
+    bool                       ownersExhausted;// all owners of the key are in triedNodes
+                                               // → further retry reaches only a proxy, or nothing
+    // NB: no stored `retriable` — futility is derived via isTransient() (§3.3, D4).
     // message() via std::runtime_error::what()
 };
 ```
 
+The exception carries **raw facts** (`phase`, `serverStatus`, `triedNodes`) plus
+one derived advisory (`ownersExhausted`). Futility (transient vs permanent) is a
+free function, not a stored verdict (D4) — see §3.3. `ownersExhausted` is
+computed for free from `consistentHash_.getOwners(key, topology_)`
+(`RemoteCache.cpp:756`): `ownersExhausted = (owners ⊆ triedNodes)`.
+
 **`phase` is the highest-value field** — more than `retriable` — because it is
 exactly what lets the user make the idempotency call:
 - **BeforeSend** — couldn't connect / `write()` never started → clean, always
@@ -112,9 +161,11 @@ Ops gain an optional retry/exclusion context, defaulted so existing call sites a
 untouched:
 
 ```cpp
-struct RetryContext {
+struct RetryContext {                          // USER→library input only
     std::vector<ServerAddress> excludeNodes;   // don't dispatch here again
     // (room to grow: deadline, attempt count, ...)
+    // NB: NOT proxyToNonOwner — that's routing policy, kept client-level
+    //     (config), matching how Java models balancing/maxRetries (D5).
 };
 
 // existing signature keeps working:
@@ -141,6 +192,78 @@ for (;;) {
 dead node(s). Retry is then just the normal op path with an exclusion set — **no
 closure capture, no shared dispatcher, no templated exception.**
 
+The existing selection (`RemoteCache.cpp:746`) already has the shape: its
+owner loop (761) and fallback "any server" loop (780) skip already-tried nodes
+(782–793). The only new code is (a) an `isExcluded(node, excludeNodes)` predicate
+in both loops, and (b) reporting the nodes the internal sweep touched so they
+land in `triedNodes`. `proxyToNonOwner` (client-level, D5) gates loop (b): when
+`false`, the owner loop failing throws `ownersExhausted=true` instead of entering
+the non-owner fallback.
+
+### 3.3 Futility classification (11a, D4)
+Raw facts on the exception; classification is a free function — advisory, never a
+safety verdict. The user ANDs it with their own idempotency judgment.
+
+```cpp
+// Server ERROR (0x50) status → transient? Mirrors Java's retriable set.
+bool isTransientStatus(uint8_t status) {
+    switch (status) {
+        case Status::NODE_SUSPECTED:          // 0x87  node-level: try another owner
+        case Status::ILLEGAL_LIFECYCLE_STATE: // 0x88  node starting/stopping
+        case Status::COMMAND_TIMEOUT:         // 0x86  busy/slow node → retry elsewhere
+            return true;                      //       (ambiguous — see outcomeUncertain)
+        case Status::INVALID_MAGIC_OR_MESSAGE_ID: // 0x81 ─┐ request-level: same
+        case Status::UNKNOWN_COMMAND:             // 0x82  │ request → same answer
+        case Status::UNKNOWN_VERSION:             // 0x83  │ on any node. Futile.
+        case Status::REQUEST_PARSING_ERROR:       // 0x84  │
+        case Status::SERVER_ERROR:                // 0x85 ─┘
+            return false;
+        default: return false;                    // unknown → treat as permanent
+    }
+}
+
+bool isTransient(const HotRodClientException& e) {   // axis 2: worth retrying?
+    switch (e.phase) {
+        case FailurePhase::BeforeSend: return true;   // nothing sent
+        case FailurePhase::AfterSend:  return true;   // not futile, but AMBIGUOUS
+        case FailurePhase::ServerError:
+            return e.serverStatus && isTransientStatus(*e.serverStatus);
+    }
+    return false;
+}
+
+// axis 1 helper: might the op have applied despite failing? ORTHOGONAL to transient.
+// The user ANDs this with their own op-idempotency knowledge before retrying.
+bool outcomeUncertain(const HotRodClientException& e) {
+    return e.phase == FailurePhase::AfterSend                    // sent, reply lost
+        || (e.phase == FailurePhase::ServerError
+            && e.serverStatus == Status::COMMAND_TIMEOUT);       // 0x86 may have applied
+}
+```
+
+**0x86 COMMAND_TIMEOUT is transient *and* uncertain** — same profile as
+`AfterSend` (a client read-timeout and a server command-timeout are the same
+situation from two sides). Retry is allowed (likely a busy node; exclusion routes
+it elsewhere) but the user must gate it on op idempotency via `outcomeUncertain`.
+**This intentionally diverges from Java**, which treats timeout as terminal
+(`HotRodTimeoutException`); our two-axis model lets us *retry-but-flag* instead.
+
+**Phase assignment lives in the transport** (network errors carry no status):
+`getConnection`/`connect` failure → `BeforeSend`; any failure once `write()` has
+*begun* (even a partial write) or while awaiting the response → `AfterSend`;
+ERROR 0x50 → `ServerError`. Rule of thumb: **write started at all ⇒ AfterSend**.
+
+**Terminal exhaustion vs permanent:** when selection runs out of nodes ("No
+servers available", `RemoteCache.cpp:809`), throw a **non-transient** exception so
+the user loop stops. That is *exhausted* ("no node could serve it now"), distinct
+from *permanent* (`0x81–0x86`, "request itself rejected") — both stop the loop,
+but the diagnosis differs.
+
+**Missing constants:** `HeaderCodec.h::Status` currently lacks `0x84`
+`REQUEST_PARSING_ERROR`, `0x87` `NODE_SUSPECTED`, `0x88`
+`ILLEGAL_LIFECYCLE_STATE` — add them (cross-check `hotrod40.ksy` + Java
+`HotRodConstants`) as part of 11a.
+
 ---
 
 ## 4. Fork 1.a vs 1.b — impact comparison (for the record)
@@ -168,44 +291,76 @@ lifetime-footgun and keeps the error trivially copyable/loggable/storable.
 
 ## 5. Progress checklist
 
-### 11a — Error surfacing (do first)
-- [ ] Add opcode/status constants for ERROR response (0x50) in `HeaderCodec.h`.
-- [ ] Parse the ERROR response body (length-prefixed error message).
-- [ ] Define `FailurePhase` + `HotRodClientException` (pure data, §3.1).
-- [ ] Detect `phase` in the transport (BeforeSend vs AfterSend around `write()`;
+### 11a — Error surfacing — ✅ shipped 2026-09-25
+- [x] Add ERROR response (0x50) + missing status constants to `HeaderCodec.h`:
+      `0x84 REQUEST_PARSING_ERROR`, `0x87 NODE_SUSPECTED`,
+      `0x88 ILLEGAL_LIFECYCLE_STATE` (§3.3).
+- [x] Parse the ERROR response body (length-prefixed error message). Drained in
+      `readLoop` *before* the pending lookup so an orphan ERROR can't desync the
+      stream.
+- [x] Define `FailurePhase` + `HotRodClientException` (pure data + `ownersExhausted`, §3.1).
+- [x] Add `isTransientStatus()` / `isTransient()` / `outcomeUncertain()` free
+      helpers (§3.3, D4). `0x86` is transient + uncertain.
+- [x] Detect `phase` in the transport (BeforeSend vs AfterSend around `write()`;
       ServerError on 0x50) and populate `triedNodes`.
-- [ ] Map response statuses → `HotRodClientException` at the existing throw sites
+- [x] Map response statuses → `HotRodClientException` at the existing throw sites
       (`RemoteCache.cpp:84,237,291,411,449,500,...`), replacing `runtime_error`.
-- [ ] Unit tests: ERROR-body parse; each phase; status→exception mapping.
-- [ ] Integration test: trigger a server ERROR and assert the typed exception.
+- [x] Make "No servers available" (`RemoteCache.cpp:809`) a non-transient exception.
+- [x] Unit tests: ERROR-body parse; each phase; status→exception mapping; isTransient.
+- [x] Integration test: trigger a server ERROR and assert the typed exception
+      (+ `ConnectionUsableAfterServerError` proves no stream desync).
 
 ### 11b — User-decided retry
 - [ ] `RetryContext` struct (§3.2), defaulted on every op signature.
 - [ ] Exclusion-aware `selectServerForKey` variant (skip `excludeNodes`).
+- [ ] Client-level `proxyToNonOwner` flag (default `true`, D5): gates the
+      owner→non-owner fallback; when `false`, throw `ownersExhausted=true`.
+- [ ] Populate `ownersExhausted` from `getOwners(key) ⊆ triedNodes`.
 - [ ] Thread-safety: guard `topology_` + `connectionPool_` + selection path
       (retry may be driven concurrently with the read-loop's topology updates).
-- [ ] `triedNodes` accumulation across successive retries (union).
-- [ ] Unit tests: exclusion selection; context threading; union accumulation.
+- [ ] `triedNodes` accumulation across successive retries (union); the internal
+      BeforeSend sweep reports *all* nodes it touched.
+- [ ] Unit tests: exclusion selection; context threading; union accumulation;
+      proxyToNonOwner on/off; ownersExhausted.
 - [ ] Integration test: kill a node mid-run, user-loop retry lands on another.
 - [ ] Docs/example: the retry loop pattern (§3.2) in `examples/quickstart/`.
 
 ---
 
 ## 6. Open questions (resolve as we implement)
-- Which Hot Rod statuses / ERROR codes are classified `retriable` and with which
-  `phase`? (Cross-check Java `exceptions.*` + `hotrod40.ksy` error response.)
+- **[RESOLVED, D4 / 2026-09-25]** Classification decided (§3.3): transient =
+  node-level `0x87`/`0x88` **and `0x86 COMMAND_TIMEOUT`** (busy node → retry
+  elsewhere, flagged `outcomeUncertain`); permanent = request-level `0x81–0x85`.
+  `0x86` intentionally diverges from Java. Only residual doubt: `0x85
+  SERVER_ERROR` stays permanent — revisit if a real case wants it retried.
 - Minimal exception vs a small subtype family — start with one type; revisit only
   if a real `catch`-differentiation need appears.
-- Interaction with the existing connection-establishment failover in
-  `selectServerForKey` — the exclusion set should compose with it, not fight it.
+- **[RESOLVED, D5]** Interaction with the connection-establishment failover in
+  `selectServerForKey`: the internal sweep is **BeforeSend-only** (it wraps
+  `getConnectionForServer`; once a live conn is returned, dispatch failures
+  propagate as AfterSend/ServerError, never internally retried). The `excludeNodes`
+  set composes with it; `proxyToNonOwner` (client-level, default `true`) gates
+  whether the sweep continues past owners into non-owners. The sweep must report
+  *every* node it touched into `triedNodes` (union) so a user retry doesn't
+  re-hit a known-dead node.
 - Do we need a `deadline`/max-attempts hint in `RetryContext`, or leave all
-  pacing to the user? (Lean: leave to the user for now.)
+  pacing to the user? (Lean: leave to the user for now. Java's automatic
+  `max_retries` default is 3 — a reference point, not a binding choice since our
+  retry is user-driven.)
 
 ---
 
 ## 7. References
 - Java behavior/semantics (authoritative): `org.infinispan.client.hotrod.exceptions.*`
   — local `/home/rigazilla/git/infinispan/client/hotrod-client/`.
+- Java routing/retry cross-check (2026-09-25): non-owner fallback in
+  `impl/transport/netty/OperationDispatcher.java` (`addressForObject` →
+  `balancer.nextServer`); default balancer `RoundRobinBalancingStrategy`;
+  `max_retries` default 3 in `impl/ConfigurationProperties.java`
+  (`DEFAULT_MAX_RETRIES`). Retriable set: `RemoteNodeSuspectException` (0x87),
+  `RemoteIllegalLifecycleStateException` (0x88), `TransportException` (no status);
+  genuine server errors terminal. Note: Java retries **automatically**; we chose
+  **user-decided** (D2) — the divergence is intentional.
 - Wire layout: `hotrod40.ksy` error response — local
   `../hotrod-dissector/schemas/`.
 - Source-hierarchy rule: [`DECISIONS.md`](DECISIONS.md) 2026-09-16 entry.
diff --git a/docs/STATUS.md b/docs/STATUS.md
index 5387d02..18ce6a5 100644
--- a/docs/STATUS.md
+++ b/docs/STATUS.md
@@ -4,7 +4,7 @@
 > If any other doc disagrees with this file, this file wins. Point-in-time
 > snapshots live in [`archive/`](archive/) and are historical only.
 >
-> **Last updated:** 2026-09-21
+> **Last updated:** 2026-09-25
 
 ---
 
@@ -33,16 +33,17 @@ _(Full per-session workflow: [`WORKFLOW.md`](WORKFLOW.md).)_
 
 _The 1–3 concrete things to do next. Keep this short and current._
 
-1. **Step 10 is complete.** `getWithMetadata` (0x1B/0x1C), `removeWithVersion`
-   (0x0D/0x0E), `replaceWithVersion` (0x09/0x0A) and the conditionals
-   `putIfAbsent` (0x05/0x06), `replace` (0x07/0x08), `containsKey` (0x0F/0x10)
-   are all **shipped** (see "Working and shipped"). Next roadmap step:
-   **Step 11 — Error handling.** Design agreed 2026-09-21, split into **11a**
-   (typed `HotRodClientException` + ERROR 0x50 parsing — do first) and **11b**
-   (user-decided retry via a defaulted `RetryContext` on the cache). Full design
-   + progress checklist: [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md);
-   rationale in [`DECISIONS.md`](DECISIONS.md) (2026-09-21 entry). **Start with
-   the 11a checklist.**
+1. **Step 11a is complete.** Typed `HotRodClientException` (pure-data:
+   `FailurePhase` + `serverStatus` + `triedNodes` + `ownersExhausted`) now
+   replaces bare `std::runtime_error` across the ops, and the server ERROR
+   response (opcode 0x50) is parsed and surfaced with its status + message (see
+   "Working and shipped"). Next roadmap step: **Step 11b — user-decided retry**
+   (defaulted `RetryContext` on the ops, exclusion-aware `selectServerForKey`,
+   `proxyToNonOwner` default true, thread-safe pool/topology). Retry policy /
+   idempotency stays the user's call. Full design + progress checklist:
+   [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md); rationale in
+   [`DECISIONS.md`](DECISIONS.md) (2026-09-21 and 2026-09-25 entries). **Start
+   with the 11b checklist.**
 2. **Benchmark the multiplexing path** — the async rewrite targets 5–10×
    concurrent throughput; this has not been measured yet.
 3. **Small cleanup:** read header "other params" when `paramCount > 0`
@@ -68,13 +69,20 @@ pull from here next. Step numbers follow
   [step-by-step plan](#plan-metadata-operations-step-10) below.
 - [ ] **Step 11 — Error handling.** Design agreed 2026-09-21 →
   [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md). Split:
-  - [ ] **11a — Error surfacing:** ERROR response parsing (opcode 0x50),
-    length-prefixed message extraction, a typed `HotRodClientException`
-    (pure-data: `retriable` + `FailurePhase` + `triedNodes` + `serverStatus`)
-    replacing bare `std::runtime_error`. Prerequisite for 11b.
+  - [x] **11a — Error surfacing** — **shipped 2026-09-25** (11 unit + 3
+    integration tests). ERROR response parsing (opcode 0x50) with
+    length-prefixed message drained in the read loop (keeps the stream in sync
+    even for orphan responses), surfaced as a typed `HotRodClientException`
+    (pure-data: `FailurePhase` {BeforeSend/AfterSend/ServerError} +
+    `serverStatus` + `triedNodes` + `ownersExhausted`) replacing bare
+    `std::runtime_error`, still catchable as `std::runtime_error`. Two-axis
+    classification helpers: `isTransient()` (futility — library's call) and
+    `outcomeUncertain()` (ambiguity — 0x86/AfterSend). `COMMAND_TIMEOUT` (0x86)
+    is transient **and** outcome-uncertain (intentionally diverges from Java).
   - [ ] **11b — User-decided retry:** defaulted `RetryContext` on the ops
-    (fork 1.b), exclusion-aware `selectServerForKey`, thread-safe pool/topology.
-    Retry policy/idempotency is the user's call, not automatic.
+    (fork 1.b), exclusion-aware `selectServerForKey`, `proxyToNonOwner` default
+    true, thread-safe pool/topology. Retry policy/idempotency is the user's
+    call, not automatic.
   Java ref: `org.infinispan.client.hotrod.exceptions.*`.
 - [ ] **Step 12 — Bulk operations.** `GET_ALL` (0x2F), `PUT_ALL` (0x2D),
   `BULK_GET` (0x1F, iterator-style).
@@ -119,6 +127,13 @@ multiplexing. See "Working and shipped" below._
 - **replace** (0x07/0x08) — store only if key present; returns
   `future<optional<EntryWithMetadata>>` (replaced entry when `previousValue` set)
 - **containsKey** (0x0F/0x10) — key-existence test; returns `future<bool>`
+- **Typed error surfacing (Step 11a):** server ERROR responses (opcode 0x50)
+  are parsed (status + length-prefixed message) and thrown as
+  `HotRodClientException` — a pure-data exception carrying `FailurePhase`,
+  `serverStatus`, `triedNodes`, `ownersExhausted`; still catchable as
+  `std::runtime_error`. Free helpers `isTransient()` / `outcomeUncertain()`
+  classify futility vs ambiguity. The ERROR body is drained in the read loop so
+  the connection stays usable after a server error (no stream desync)
 - **Async API:** all operations return `std::future` / `std::optional`
 - **MultiplexedConnection:** true async — dedicated read-loop thread,
   `messageId → promise` pending map, `execute()` used by all four operations.
@@ -130,14 +145,15 @@ multiplexing. See "Working and shipped" below._
   Docker daemon; Windows doesn't build the integration tests). Windows/MSVC
   portability and `-Werror` build parity also landed (Sept 2026).
 
-**Test status (verified 2026-09-20):**
-- Unit: **176/176** passing (`./build/unit_tests`, <1s)
-- Integration: **75/75** passing across 14 suites (`ctest`, spins up Docker
+**Test status (verified 2026-09-25):**
+- Unit: **187/187** passing (`./build/unit_tests`, <1s)
+- Integration: **78/78** passing across 15 suites (`ctest`, spins up Docker
   Infinispan single-server + multi-node clusters), now also green on Linux CI.
   The interlaced distributed tests (`ConcurrentMultiServerTest`) were fixed to
   tolerate a GET racing ahead of its PUT — a `nullopt` is expected, only a
   present-but-wrong value is an error.
-- Full run: `ctest --test-dir build --output-on-failure` → 100% pass (10 ctest targets)
+- Full run: `ctest --test-dir build --output-on-failure` → 100% pass
+  (16 ctest tests: 1 unit + 15 integration suites)
 - **Local caveat:** the multi-node cluster suites bind fixed host ports
   `11222/11322/11422/11522`; free `11222` (e.g. stop the `memory-service`
   Infinispan) before running them locally, or they fail with "port is already
diff --git a/include/hotrod/HeaderCodec.h b/include/hotrod/HeaderCodec.h
index 469a1db..10a2af7 100644
--- a/include/hotrod/HeaderCodec.h
+++ b/include/hotrod/HeaderCodec.h
@@ -64,8 +64,11 @@ namespace Status {
     const uint8_t INVALID_MAGIC_OR_MESSAGE_ID = 0x81;
     const uint8_t UNKNOWN_COMMAND = 0x82;
     const uint8_t UNKNOWN_VERSION = 0x83;
+    const uint8_t REQUEST_PARSING_ERROR = 0x84;
     const uint8_t SERVER_ERROR = 0x85;
     const uint8_t COMMAND_TIMEOUT = 0x86;
+    const uint8_t NODE_SUSPECTED = 0x87;
+    const uint8_t ILLEGAL_LIFECYCLE_STATE = 0x88;
 }
 
 /**
diff --git a/include/hotrod/HotRodClientException.h b/include/hotrod/HotRodClientException.h
new file mode 100644
index 0000000..3f57a06
--- /dev/null
+++ b/include/hotrod/HotRodClientException.h
@@ -0,0 +1,115 @@
+#pragma once
+
+#include "Types.h"
+#include "HeaderCodec.h"  // Status:: constants
+#include <stdexcept>
+#include <string>
+#include <vector>
+#include <optional>
+#include <cstdint>
+
+namespace hotrod {
+
+/**
+ * Where an operation failed. This is the highest-value field on
+ * HotRodClientException: it is what lets a caller decide whether retrying THIS
+ * operation is safe (idempotency), independent of whether a retry is worthwhile.
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.1 / D4.
+ */
+enum class FailurePhase {
+    BeforeSend,   // request never left the client (connect/write not started) — safe to retry
+    AfterSend,    // request was (partially) sent, no usable response — AMBIGUOUS (may have applied)
+    ServerError   // server replied with an ERROR (0x50) status
+};
+
+/**
+ * Minimal, pure-value address of a server an operation was dispatched to.
+ * Kept independent of ServerInfo/TopologyInfo so the exception stays trivially
+ * copyable and free of client-internal coupling.
+ */
+struct ServerAddress {
+    std::string host;
+    uint16_t    port = 0;
+
+    bool operator==(const ServerAddress& other) const {
+        return host == other.host && port == other.port;
+    }
+};
+
+/**
+ * Typed exception for all Hot Rod client failures (Step 11a).
+ *
+ * Pure data: it carries the raw facts about a failure (phase, server status,
+ * nodes tried) and nothing that captures client internals — so it is safe to
+ * copy, log, and store. Whether a retry is *worthwhile* (isTransient) or the
+ * outcome is *uncertain* (outcomeUncertain) is DERIVED via the free helpers
+ * below, never baked in as a stored verdict.
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.1 / §3.3.
+ */
+class HotRodClientException : public std::runtime_error {
+public:
+    FailurePhase               phase;
+    std::optional<uint8_t>     serverStatus;         // Hot Rod status / ERROR code, if any
+    std::vector<ServerAddress> triedNodes;           // nodes attempted (union across retries)
+    bool                       ownersExhausted;      // all owners of the key are in triedNodes
+
+    HotRodClientException(const std::string& message,
+                          FailurePhase phase,
+                          std::optional<uint8_t> serverStatus = std::nullopt,
+                          std::vector<ServerAddress> triedNodes = {},
+                          bool ownersExhausted = false)
+        : std::runtime_error(message)
+        , phase(phase)
+        , serverStatus(serverStatus)
+        , triedNodes(std::move(triedNodes))
+        , ownersExhausted(ownersExhausted)
+    {}
+};
+
+// --- Classification helpers (advisory; see D4). Free functions, not verdicts. ---
+
+/**
+ * Is a server ERROR status worth retrying (i.e. NOT futile)?
+ * Node-level failures (0x86/0x87/0x88) can clear on another node; request-level
+ * failures (0x81–0x85) reproduce identically anywhere, so they are permanent.
+ */
+inline bool isTransientStatus(uint8_t status) {
+    switch (status) {
+        case Status::COMMAND_TIMEOUT:         // 0x86 busy/slow node (also outcome-uncertain)
+        case Status::NODE_SUSPECTED:          // 0x87 node suspected
+        case Status::ILLEGAL_LIFECYCLE_STATE: // 0x88 node starting/stopping
+            return true;
+        default:
+            return false;                     // 0x81..0x85 and unknown → permanent
+    }
+}
+
+/**
+ * Axis 2 — "could a retry plausibly succeed?". Advisory only; the caller must
+ * AND this with their own knowledge of whether THIS op is safe to retry
+ * (see outcomeUncertain).
+ */
+inline bool isTransient(const HotRodClientException& e) {
+    switch (e.phase) {
+        case FailurePhase::BeforeSend: return true;   // nothing sent
+        case FailurePhase::AfterSend:  return true;   // not futile, but ambiguous
+        case FailurePhase::ServerError:
+            return e.serverStatus.has_value() && isTransientStatus(*e.serverStatus);
+    }
+    return false;
+}
+
+/**
+ * Axis 1 helper — "might the operation have applied despite the failure?".
+ * Orthogonal to isTransient(). A command timeout is transient AND uncertain,
+ * the same profile as a mid-flight drop (AfterSend).
+ */
+inline bool outcomeUncertain(const HotRodClientException& e) {
+    return e.phase == FailurePhase::AfterSend
+        || (e.phase == FailurePhase::ServerError &&
+            e.serverStatus.has_value() && *e.serverStatus == Status::COMMAND_TIMEOUT);
+}
+
+} // namespace hotrod
diff --git a/src/operations/RemoteCache.cpp b/src/operations/RemoteCache.cpp
index 1b1fc61..efe032f 100644
--- a/src/operations/RemoteCache.cpp
+++ b/src/operations/RemoteCache.cpp
@@ -1,5 +1,6 @@
 #include "hotrod/RemoteCache.h"
 #include "hotrod/Codec.h"
+#include "hotrod/HotRodClientException.h"
 #include <stdexcept>
 #include <set>
 
@@ -81,7 +82,8 @@ namespace hotrod
       {
          if (status != 0x00)
          {
-            throw std::runtime_error("PING failed with status: " + std::to_string(status));
+            throw HotRodClientException("PING failed with status: " + std::to_string(status),
+                                        FailurePhase::ServerError, status);
          }
          // PING response body (Protocol 3.0+):
          // - key_type (media_type)
@@ -234,7 +236,8 @@ namespace hotrod
          }
          if (status != 0x00)
          {
-            throw std::runtime_error("GET failed with status: " + std::to_string(status));
+            throw HotRodClientException("GET failed with status: " + std::to_string(status),
+                                        FailurePhase::ServerError, status);
          }
 
          // Read value (lp_bytes: vInt length + bytes)
@@ -288,7 +291,8 @@ namespace hotrod
          }
          if (status != 0x00)
          {
-            throw std::runtime_error("GET_WITH_METADATA failed with status: " + std::to_string(status));
+            throw HotRodClientException("GET_WITH_METADATA failed with status: " + std::to_string(status),
+                                        FailurePhase::ServerError, status);
          }
 
          // On success the body is: entry_metadata (flag + optional expiration + version)
@@ -408,7 +412,8 @@ namespace hotrod
                            if (resp.error)
                               std::rethrow_exception(resp.error);
                            if (resp.status != 0x00 && resp.status != 0x03)
-                              throw std::runtime_error("PUT failed with status: " + std::to_string(resp.status));
+                              throw HotRodClientException("PUT failed with status: " + std::to_string(resp.status),
+                                                         FailurePhase::ServerError, resp.status);
                            if (resp.status == 0x03 && resp.body.has_value()) {
                               return std::any_cast<EntryWithMetadata>(resp.body);
                            }
@@ -446,7 +451,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("REMOVE failed with status: " + std::to_string(status));
+         throw HotRodClientException("REMOVE failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       // Execute
@@ -497,7 +503,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("REMOVE_WITH_VERSION failed with status: " + std::to_string(status));
+         throw HotRodClientException("REMOVE_WITH_VERSION failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       // Execute
@@ -568,7 +575,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("REPLACE_WITH_VERSION failed with status: " + std::to_string(status));
+         throw HotRodClientException("REPLACE_WITH_VERSION failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       // Execute
@@ -629,7 +637,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("PUT_IF_ABSENT failed with status: " + std::to_string(status));
+         throw HotRodClientException("PUT_IF_ABSENT failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       MultiplexedConnection *conn = selectServerForKey(key);
@@ -689,7 +698,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("REPLACE failed with status: " + std::to_string(status));
+         throw HotRodClientException("REPLACE failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       MultiplexedConnection *conn = selectServerForKey(key);
@@ -726,7 +736,8 @@ namespace hotrod
          {
             return {};
          }
-         throw std::runtime_error("CONTAINS_KEY failed with status: " + std::to_string(status));
+         throw HotRodClientException("CONTAINS_KEY failed with status: " + std::to_string(status),
+                                     FailurePhase::ServerError, status);
       };
 
       MultiplexedConnection *conn = selectServerForKey(key);
@@ -805,8 +816,13 @@ namespace hotrod
                }
             }
 
-            // No servers available
-            throw std::runtime_error("No servers available in topology");
+            // No servers available: every owner and fallback server was
+            // unreachable. Nothing was sent (BeforeSend); ownersExhausted marks
+            // that all owners were tried. (The retry loop that treats exhaustion
+            // as terminal/non-transient lands in 11b.)
+            throw HotRodClientException("No servers available in topology",
+                                        FailurePhase::BeforeSend, std::nullopt,
+                                        {}, /*ownersExhausted=*/true);
          }
          else
          {
diff --git a/src/transport/MultiplexedConnection.cpp b/src/transport/MultiplexedConnection.cpp
index 56d86d1..e28ced9 100644
--- a/src/transport/MultiplexedConnection.cpp
+++ b/src/transport/MultiplexedConnection.cpp
@@ -1,6 +1,7 @@
 #include "hotrod/MultiplexedConnection.h"
 #include "hotrod/Connection.h"
 #include "hotrod/HeaderCodec.h"
+#include "hotrod/HotRodClientException.h"
 #include "hotrod/Codec.h"
 #include <stdexcept>
 #include <iostream>
@@ -95,7 +96,9 @@ std::future<Response> MultiplexedConnection::execute(
     int32_t flags)
 {
     if (!connected_) {
-        throw std::runtime_error("Not connected");
+        // Nothing has been sent — always safe to retry (any op).
+        throw HotRodClientException("Not connected", FailurePhase::BeforeSend,
+                                    std::nullopt, {{host_, port_}});
     }
 
     // Generate unique message ID
@@ -122,13 +125,17 @@ std::future<Response> MultiplexedConnection::execute(
     try {
         std::lock_guard<std::mutex> lock(writeMutex_);
         connection_->send(completeRequest);
-    } catch (...) {
+    } catch (const std::exception& e) {
         // Remove pending on write error
         {
             std::lock_guard<std::mutex> lock(pendingMutex_);
             pending_.erase(msgId);
         }
-        throw;
+        // A partial write may already have reached the server — be conservative
+        // and treat this as AfterSend (ambiguous), not BeforeSend.
+        throw HotRodClientException(std::string("Send failed: ") + e.what(),
+                                    FailurePhase::AfterSend, std::nullopt,
+                                    {{host_, port_}});
     }
 
     return future;
@@ -173,6 +180,14 @@ void MultiplexedConnection::readLoop() {
                 topologyUpdate = std::move(topo);
             }
 
+            // 1b. Server ERROR response (0x50): the body is a length-prefixed
+            // error message. Read it NOW — before the pending lookup — so the
+            // stream stays in sync even if the request has no waiter (orphan).
+            std::optional<std::string> errorMessage;
+            if (opcode == Opcode::ERROR_RESPONSE) {
+                errorMessage = connection_->receiveString();
+            }
+
             // 2. Find pending request
             std::unique_ptr<PendingRequest> pending;
             {
@@ -192,13 +207,30 @@ void MultiplexedConnection::readLoop() {
                 onTopologyUpdate_(*topologyUpdate, pending->cacheName);
             }
 
+            // 2b. Deliver server ERROR (0x50) as a typed exception, skipping the
+            // opcode/body checks below (the ERROR opcode never matches expected).
+            if (errorMessage) {
+                Response resp;
+                resp.status = status;
+                resp.error = std::make_exception_ptr(HotRodClientException(
+                    "Server error (status " + std::to_string(status) + "): " + *errorMessage,
+                    FailurePhase::ServerError, status, {{host_, port_}}));
+                resp.topologyUpdate = std::move(topologyUpdate);
+                pending->promise.set_value(std::move(resp));
+                continue;
+            }
+
             // 3. Validate opcode
             if (opcode != pending->expectedOpcode) {
                 Response resp;
-                resp.error = std::make_exception_ptr(
-                    std::runtime_error("Opcode mismatch: expected " +
-                                     std::to_string(pending->expectedOpcode) +
-                                     ", got " + std::to_string(opcode)));
+                // Protocol-level mismatch: we got a reply, but the wrong one.
+                // Not a server ERROR status, and not worth retrying → ServerError
+                // phase with no status makes isTransient() false.
+                resp.error = std::make_exception_ptr(HotRodClientException(
+                    "Opcode mismatch: expected " +
+                        std::to_string(pending->expectedOpcode) +
+                        ", got " + std::to_string(opcode),
+                    FailurePhase::ServerError, std::nullopt, {{host_, port_}}));
                 pending->promise.set_value(std::move(resp));
                 continue;
             }
@@ -268,7 +300,10 @@ ByteArray MultiplexedConnection::buildRequest(
 
 void MultiplexedConnection::closeAllPending(const std::string& errorMsg) {
     std::lock_guard<std::mutex> lock(pendingMutex_);
-    auto error = std::make_exception_ptr(std::runtime_error(errorMsg));
+    // These requests were already sent; the connection dropped before a response
+    // arrived → AfterSend (ambiguous, may have applied on the server).
+    auto error = std::make_exception_ptr(HotRodClientException(
+        errorMsg, FailurePhase::AfterSend, std::nullopt, {{host_, port_}}));
 
     for (auto& [msgId, pending] : pending_) {
         Response resp;
diff --git a/tests/integration/ErrorHandlingIntegrationTest.cpp b/tests/integration/ErrorHandlingIntegrationTest.cpp
new file mode 100644
index 0000000..132f577
--- /dev/null
+++ b/tests/integration/ErrorHandlingIntegrationTest.cpp
@@ -0,0 +1,100 @@
+#include <tuple>
+#include <gtest/gtest.h>
+#include "hotrod/RemoteCache.h"
+#include "hotrod/HotRodClientException.h"
+#include "InfinispanTestEnvironment.h"
+#include <cstdlib>
+#include <string>
+
+using namespace hotrod;
+using namespace hotrod::test;
+
+namespace {
+// Create a cache via the server CLI (same approach as the other integration tests).
+void createCacheViaCLI(const std::string& cacheName) {
+    std::string cmd = "docker exec " + InfinispanTestEnvironment::containerID +
+                     " bash -c \"echo 'create cache --template=org.infinispan.DIST_SYNC " + cacheName +
+                     "' | /opt/infinispan/bin/cli.sh -c http://admin:password@localhost:11222\" >/dev/null 2>&1";
+    std::ignore = system(cmd.c_str());
+}
+} // namespace
+
+/**
+ * Step 11a — Error handling integration tests.
+ *
+ * Exercises the real server ERROR (0x50) path end-to-end: an operation against a
+ * cache that does not exist makes the server reply with an ERROR response, which
+ * the client must surface as a typed HotRodClientException carrying the server
+ * status and message — not desync the connection.
+ *
+ * Reference: docs/ERROR_HANDLING_DESIGN.md §3.1.
+ */
+
+// A cache name that is never created → server returns an ERROR response.
+static const char* kMissingCache = "no_such_cache_11a";
+
+// Operating on an undefined cache surfaces a typed HotRodClientException.
+TEST(ErrorHandlingIntegrationTest, ServerErrorOnMissingCache) {
+    RemoteCache cache(InfinispanTestEnvironment::host,
+                      InfinispanTestEnvironment::port,
+                      kMissingCache);
+    cache.connect();
+
+    ByteArray key = {'k', 'e', 'y'};
+
+    try {
+        cache.get(key).get();
+        FAIL() << "expected a HotRodClientException for a missing cache";
+    } catch (const HotRodClientException& e) {
+        EXPECT_EQ(FailurePhase::ServerError, e.phase);
+        EXPECT_TRUE(e.serverStatus.has_value());
+        EXPECT_STRNE("", e.what());  // server-provided message propagated
+    }
+
+    cache.disconnect();
+}
+
+// The typed exception is still catchable as std::runtime_error (back-compat).
+TEST(ErrorHandlingIntegrationTest, ServerErrorCatchableAsRuntimeError) {
+    RemoteCache cache(InfinispanTestEnvironment::host,
+                      InfinispanTestEnvironment::port,
+                      kMissingCache);
+    cache.connect();
+
+    ByteArray key = {'k'};
+    EXPECT_THROW({ cache.get(key).get(); }, std::runtime_error);
+
+    cache.disconnect();
+}
+
+// After a server ERROR the connection is still usable — the ERROR body was fully
+// drained, so the stream stayed in sync. A real round-trip on the same connection
+// against a valid cache must succeed (a desync would hang or return garbage).
+TEST(ErrorHandlingIntegrationTest, ConnectionUsableAfterServerError) {
+    createCacheViaCLI("err_ok_cache");
+
+    RemoteCache cache(InfinispanTestEnvironment::host,
+                      InfinispanTestEnvironment::port,
+                      kMissingCache);
+    cache.connect();
+
+    ByteArray key = {'k'};
+    EXPECT_THROW({ cache.get(key).get(); }, HotRodClientException);
+
+    // Same connection, now target the valid cache: PUT then GET must round-trip.
+    cache.setCacheName("err_ok_cache");
+    ByteArray value = {'v'};
+    ASSERT_NO_THROW({ cache.put(key, value).get(); });
+    auto got = cache.get(key).get();
+    ASSERT_TRUE(got.has_value());
+    EXPECT_EQ(value, *got);
+
+    cache.disconnect();
+}
+
+// Main function - registers the global environment
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    ::testing::AddGlobalTestEnvironment(new InfinispanTestEnvironment());
+    return RUN_ALL_TESTS();
+}
diff --git a/tests/unit/ErrorHandlingTest.cpp b/tests/unit/ErrorHandlingTest.cpp
new file mode 100644
index 0000000..92a9327
--- /dev/null
+++ b/tests/unit/ErrorHandlingTest.cpp
@@ -0,0 +1,166 @@
+#include <gtest/gtest.h>
+#include "hotrod/HotRodClientException.h"
+#include "hotrod/HeaderCodec.h"
+#include "hotrod/Codec.h"
+#include <vector>
+#include <string>
+
+using namespace hotrod;
+
+/**
+ * Step 11a — Error handling unit tests.
+ *
+ * Covers:
+ * - The typed HotRodClientException (pure-data fields).
+ * - Futility classification: isTransientStatus / isTransient (D4, §3.3).
+ * - Ambiguity: outcomeUncertain (axis 1 helper).
+ * - Wire decoding of the ERROR response (opcode 0x50 + length-prefixed message).
+ *
+ * Reference: docs/ERROR_HANDLING_DESIGN.md §3.1, §3.3.
+ */
+
+// --- Status constants -------------------------------------------------------
+
+TEST(ErrorHandlingTest, StatusConstantsMatchProtocol) {
+    EXPECT_EQ(0x81, Status::INVALID_MAGIC_OR_MESSAGE_ID);
+    EXPECT_EQ(0x82, Status::UNKNOWN_COMMAND);
+    EXPECT_EQ(0x83, Status::UNKNOWN_VERSION);
+    EXPECT_EQ(0x84, Status::REQUEST_PARSING_ERROR);
+    EXPECT_EQ(0x85, Status::SERVER_ERROR);
+    EXPECT_EQ(0x86, Status::COMMAND_TIMEOUT);
+    EXPECT_EQ(0x87, Status::NODE_SUSPECTED);
+    EXPECT_EQ(0x88, Status::ILLEGAL_LIFECYCLE_STATE);
+    EXPECT_EQ(0x50, Opcode::ERROR_RESPONSE);
+}
+
+// --- Exception shape --------------------------------------------------------
+
+TEST(ErrorHandlingTest, ExceptionCarriesRawFacts) {
+    HotRodClientException e("boom", FailurePhase::ServerError, Status::SERVER_ERROR,
+                            {{"node-a", 11222}, {"node-b", 11322}}, /*ownersExhausted=*/true);
+
+    EXPECT_STREQ("boom", e.what());
+    EXPECT_EQ(FailurePhase::ServerError, e.phase);
+    ASSERT_TRUE(e.serverStatus.has_value());
+    EXPECT_EQ(Status::SERVER_ERROR, *e.serverStatus);
+    ASSERT_EQ(2u, e.triedNodes.size());
+    EXPECT_EQ("node-a", e.triedNodes[0].host);
+    EXPECT_EQ(11222, e.triedNodes[0].port);
+    EXPECT_TRUE(e.ownersExhausted);
+}
+
+TEST(ErrorHandlingTest, ExceptionDefaults) {
+    HotRodClientException e("x", FailurePhase::BeforeSend);
+    EXPECT_FALSE(e.serverStatus.has_value());
+    EXPECT_TRUE(e.triedNodes.empty());
+    EXPECT_FALSE(e.ownersExhausted);
+}
+
+TEST(ErrorHandlingTest, IsCatchableAsStdException) {
+    try {
+        throw HotRodClientException("x", FailurePhase::AfterSend);
+    } catch (const std::runtime_error& e) {
+        EXPECT_STREQ("x", e.what());
+        return;
+    }
+    FAIL() << "expected to be caught as std::runtime_error";
+}
+
+// --- isTransientStatus (futility of a server ERROR code) --------------------
+
+TEST(ErrorHandlingTest, TransientStatuses) {
+    // Node-level + command timeout → retry may succeed elsewhere.
+    EXPECT_TRUE(isTransientStatus(Status::COMMAND_TIMEOUT));         // 0x86
+    EXPECT_TRUE(isTransientStatus(Status::NODE_SUSPECTED));          // 0x87
+    EXPECT_TRUE(isTransientStatus(Status::ILLEGAL_LIFECYCLE_STATE)); // 0x88
+}
+
+TEST(ErrorHandlingTest, PermanentStatuses) {
+    // Request-level: identical request → identical failure anywhere. Futile.
+    EXPECT_FALSE(isTransientStatus(Status::INVALID_MAGIC_OR_MESSAGE_ID)); // 0x81
+    EXPECT_FALSE(isTransientStatus(Status::UNKNOWN_COMMAND));             // 0x82
+    EXPECT_FALSE(isTransientStatus(Status::UNKNOWN_VERSION));             // 0x83
+    EXPECT_FALSE(isTransientStatus(Status::REQUEST_PARSING_ERROR));       // 0x84
+    EXPECT_FALSE(isTransientStatus(Status::SERVER_ERROR));                // 0x85
+    EXPECT_FALSE(isTransientStatus(0xFF));                                // unknown
+}
+
+// --- isTransient (per phase) ------------------------------------------------
+
+TEST(ErrorHandlingTest, IsTransientByPhase) {
+    EXPECT_TRUE(isTransient(HotRodClientException("x", FailurePhase::BeforeSend)));
+    EXPECT_TRUE(isTransient(HotRodClientException("x", FailurePhase::AfterSend)));
+
+    // ServerError classified by status.
+    EXPECT_TRUE(isTransient(HotRodClientException(
+        "x", FailurePhase::ServerError, Status::NODE_SUSPECTED)));
+    EXPECT_TRUE(isTransient(HotRodClientException(
+        "x", FailurePhase::ServerError, Status::COMMAND_TIMEOUT)));
+    EXPECT_FALSE(isTransient(HotRodClientException(
+        "x", FailurePhase::ServerError, Status::SERVER_ERROR)));
+
+    // ServerError with no status (e.g. opcode mismatch) → not transient.
+    EXPECT_FALSE(isTransient(HotRodClientException("x", FailurePhase::ServerError)));
+}
+
+// --- outcomeUncertain (ambiguity, axis 1) -----------------------------------
+
+TEST(ErrorHandlingTest, OutcomeUncertain) {
+    // AfterSend: sent but no usable reply → may have applied.
+    EXPECT_TRUE(outcomeUncertain(HotRodClientException("x", FailurePhase::AfterSend)));
+    // Server command timeout: server may have applied before giving up.
+    EXPECT_TRUE(outcomeUncertain(HotRodClientException(
+        "x", FailurePhase::ServerError, Status::COMMAND_TIMEOUT)));
+
+    // Clean before-send or a definite server rejection: NOT uncertain.
+    EXPECT_FALSE(outcomeUncertain(HotRodClientException("x", FailurePhase::BeforeSend)));
+    EXPECT_FALSE(outcomeUncertain(HotRodClientException(
+        "x", FailurePhase::ServerError, Status::SERVER_ERROR)));
+}
+
+TEST(ErrorHandlingTest, TimeoutIsTransientAndUncertain) {
+    // 0x86 has the same profile as AfterSend: retry-allowed but ambiguous.
+    HotRodClientException e("timeout", FailurePhase::ServerError, Status::COMMAND_TIMEOUT);
+    EXPECT_TRUE(isTransient(e));
+    EXPECT_TRUE(outcomeUncertain(e));
+}
+
+// --- ERROR response wire decoding (opcode 0x50) -----------------------------
+
+TEST(ErrorHandlingTest, ParseErrorResponseBody) {
+    // Craft a server ERROR response:
+    //   Magic 0xA1 · msgId 1 · opcode 0x50 · status 0x85 · topoMarker 0x00
+    //   error message (lp_string): len + UTF-8 bytes
+    const std::string msg = "java.lang.NullPointerException";
+    ByteArray response = {
+        0xA1,        // Magic
+        0x01,        // Message ID = 1
+        0x50,        // Opcode = ERROR_RESPONSE
+        0x85,        // Status = SERVER_ERROR
+        0x00         // Topology change = none
+    };
+    Codec::writeString(response, msg);  // lp_string body
+
+    size_t offset = 0;
+    ResponseHeader header = HeaderCodec::readResponseHeader(response, offset);
+    EXPECT_EQ(Opcode::ERROR_RESPONSE, header.opcode);
+    EXPECT_EQ(Status::SERVER_ERROR, header.status);
+
+    std::string decoded = Codec::readString(response, offset);
+    EXPECT_EQ(msg, decoded);
+    EXPECT_EQ(response.size(), offset);  // whole body consumed → stream stays in sync
+}
+
+TEST(ErrorHandlingTest, ParseErrorResponseEmptyMessage) {
+    ByteArray response = {0xA1, 0x02, 0x50, 0x86, 0x00};
+    Codec::writeString(response, "");  // empty message
+
+    size_t offset = 0;
+    ResponseHeader header = HeaderCodec::readResponseHeader(response, offset);
+    EXPECT_EQ(Opcode::ERROR_RESPONSE, header.opcode);
+    EXPECT_EQ(Status::COMMAND_TIMEOUT, header.status);
+
+    std::string decoded = Codec::readString(response, offset);
+    EXPECT_TRUE(decoded.empty());
+    EXPECT_EQ(response.size(), offset);
+}