diff --git a/.add/state.json b/.add/state.json
index 6821d819..f046efb2 100644
--- a/.add/state.json
+++ b/.add/state.json
@@ -1,7 +1,7 @@
 {
   "project": "moon",
   "stage": "production",
-  "active_task": "client-identity-introspection",
+  "active_task": "protocol-error-lifetime",
   "active_milestone": "v0-9-client-compat",
   "tasks": {
     "hotpath-lock-quickwins": {
@@ -332,14 +332,14 @@
     },
     "protocol-error-lifetime": {
       "title": "Protocol errors reply and close cleanly, never stall or eat the valid prefix",
-      "phase": "ground",
+      "phase": "tests",
       "gate": "none",
       "milestone": "v0-9-client-compat",
       "depends_on": [
         "client-compat-harness"
       ],
       "created": "2026-08-09T16:42:35+00:00",
-      "updated": "2026-08-09T16:42:35+00:00"
+      "updated": "2026-08-12T04:55:05+00:00"
     },
     "multi-exec-queue-semantics": {
       "title": "MULTI queues every command, and EXEC answers correctly on every shard count",
@@ -476,7 +476,7 @@
     }
   },
   "created": "2026-06-11T03:18:21+00:00",
-  "updated": "2026-08-11T18:20:42+00:00",
+  "updated": "2026-08-12T04:55:06+00:00",
   "setup": {
     "locked": true,
     "locked_at": "2026-06-11T03:28:00+00:00",
diff --git a/.add/tasks/multi-exec-queue-semantics/TASK.md b/.add/tasks/multi-exec-queue-semantics/TASK.md
index 81d0fa0f..20f3db08 100644
--- a/.add/tasks/multi-exec-queue-semantics/TASK.md
+++ b/.add/tasks/multi-exec-queue-semantics/TASK.md
@@ -15,33 +15,131 @@ phase: ground   <!-- ground -> specify -> scenarios -> contract -> tests -> buil
 
 ## 0 · GROUND — the real codebase ▸ docs/02-the-flow.md
 
-Touches (files · symbols · signatures): <path:symbol — what it is / how it is keyed>
-Context (working folder): <docs · todos · config · data the task touches — task-delta only>
-Honors (patterns / conventions): <PROJECT.md / CONVENTIONS.md anchors — task-delta only, never a re-scan>
-Anchors the contract cites: <the symbols §3 will name>
+Touches (files · symbols · signatures):
+- The MULTI queueing step in each handler — `handler_monoio/mod.rs` (~1826, `if conn.in_multi {`),
+  `handler_sharded/mod.rs`, `handler_single.rs` (~1752). Today it pushes EVERY frame onto
+  `conn.command_queue` and answers `+QUEUED` without looking at the command at all. The only
+  exception already there is an `FT.*` rejection.
+- `src/command/metadata.rs:COMMAND_META` — `lookup()` + `arity`, which is the information the
+  queue step needs and does not consult.
+- `ConnectionState.command_queue` / `in_multi` (`src/server/conn/core.rs`), and the EXEC replay in
+  `handler_monoio/write.rs` + `src/shard/spsc_handler.rs` (the two per-command txn loops).
+
+Context (working folder): `tests/multi_queues_inline_get.rs` (guards that commands QUEUE at all —
+the #457 regression), `tests/watch_cas_transactions.rs` (WATCH/CAS), `scripts/client-compat/
+manifest.yaml`.
+
+Honors: `Frame::Error(Bytes)` for errors; per-shard locks; no allocation on the dispatch path.
+
+Anchors the contract cites: the `in_multi` queue step, `COMMAND_META::lookup`, `arity`.
+
+### Measured against redis-server 8.6.1 (Moon single-shard, identical bytes, one command per write)
+
+Probe: `scratchpad/probe_ground.py` plus stepwise follow-ups. **Cases are FLUSHALL-isolated** —
+without that, one real finding manufactured two false ones (see below).
+
+| sequence | redis-server 8.6.1 | Moon | verdict |
+|---|---|---|---|
+| `MULTI` / `NOSUCHCMD a b` / `EXEC` | `+OK` · `-ERR unknown command …'a' 'b'` · `-EXECABORT Transaction discarded because of previous errors.` | `+OK` · **`+QUEUED`** · `*1[-ERR unknown command]` | diverges |
+| `MULTI` / `GET` (bad arity) / `EXEC` | `+OK` · `-ERR wrong number of arguments for 'get' command` · `-EXECABORT` | `+OK` · **`+QUEUED`** · `*1[-ERR wrong number…]` | diverges |
+| `MULTI` / `NOSUCHCMD` / `SET k v` / `EXEC` | whole transaction DISCARDED — `SET` never runs | **`*2` — the `SET` IS APPLIED** | diverges |
+| `MULTI` / `SUBSCRIBE ch` / `EXEC` | `+QUEUED`, then EXEC runs it | **executes SUBSCRIBE immediately**, then EXEC errors | diverges |
+| `MULTI` / `*-9` (bad frame) / `EXEC` | bad frame ignored; `EXEC` → `*0` | **connection CLOSED / RST** | diverges |
+| `MULTI` / `BLPOP nokey 0` / `EXEC` | `*1` `*-1` (Null **Array**) | `*1` `$-1` (Null **Bulk**) | diverges |
+| nested `MULTI` | `-ERR MULTI calls can not be nested`, `EXEC` → `*0` | identical | matches |
+| `EXEC` / `DISCARD` without MULTI | `-ERR EXEC without MULTI` / `-ERR DISCARD without MULTI` | identical | matches |
+| `MULTI` / `SET k v` / `DISCARD` / `GET k` | `$-1` — discarded | identical | matches |
+| `MULTI` / `SELECT 3` / `EXEC` | `+OK` inside `*1` | identical | matches |
+| empty `MULTI` / `EXEC` | `*0` | identical | matches |
+| `MULTI` / `SET k v` / `RESET` / `EXEC` | `+RESET` then `-ERR EXEC without MULTI` | identical | matches |
+| WRONGTYPE inside MULTI | `*1[-WRONGTYPE …]` | identical | matches |
+
+**The headline is atomicity, not the error text.** Moon accepts anything into the queue, so a
+transaction containing ONE unrunnable command still commits every OTHER command in it. Redis
+refuses the command at queue time and then refuses the whole transaction. A client that typos one
+command gets a partial write from Moon and nothing from Redis — and it cannot tell, because Moon
+answered `+QUEUED` to the typo.
+
+**Method note (cost paid, recorded so it is not paid twice):** the first probe pass shared server
+state across cases. The missing EXECABORT applied `SET k v` on Moon only, which then made `DISCARD`
+look broken (`GET k` → `v` instead of `$-1`) and `DEL` look wrong (`:1` vs `:0`). Both were FALSE —
+verified against a fresh server, DISCARD and DEL are correct. One real bug manufactured two
+phantom ones. A second pass under host load showed empty replies that were really 2s-timeout
+truncation, not silence. Every row above was re-confirmed stepwise on an isolated server.
 
 ---
 
 ## 1 · SPECIFY — the rules ▸ docs/03-step-1-specify.md
 
-Feature: <name>
-Framings weighed: <chosen> (chosen) · <alternative> · <alternative>
+Feature: a transaction is all-or-nothing at QUEUE time, the way Redis makes it. A command that
+could never have run — unknown name, impossible arity — poisons the transaction when it is QUEUED,
+so EXEC refuses the whole block instead of applying the half that happened to be valid.
+
+Framings weighed:
+- **Validate at queue time, latch the failure, refuse at EXEC (chosen).** Redis's actual design:
+  `MULTI` opens the queue; each queued command is checked for existence and arity BEFORE it is
+  stored; a failure replies an error immediately AND sets a dirty flag; `EXEC` sees the flag and
+  answers `-EXECABORT`. Cheap (the check is a `COMMAND_META` lookup Moon already performs on every
+  dispatch) and it is the semantics every client library is written against.
+- *Validate at EXEC time by walking the queue.* Rejected: it gives the right EXEC answer but the
+  WRONG queue answer — the client is told `+QUEUED` for a command that cannot exist, so a driver
+  that surfaces per-command errors (node-redis, lettuce) reports success for the typo and only
+  fails later, at a call site that has no idea which command was bad.
+- *Do nothing; document the divergence.* Rejected: this is not cosmetic. Measured today,
+  `MULTI / NOSUCHCMD / SET k v / EXEC` leaves `k` SET on Moon and unset on Redis. A transaction
+  that partially applies is a data-correctness bug, not a compatibility nit.
+
 Must:
 <must>
-  - <required behavior>
+  - A command queued inside MULTI is checked for (a) existence and (b) arity against the same
+    `COMMAND_META` table dispatch uses. Failure replies the error immediately, does NOT queue it,
+    and marks the transaction dirty.
+  - `EXEC` on a dirty transaction replies
+    `-EXECABORT Transaction discarded because of previous errors.`, executes NOTHING, and leaves
+    the connection in the normal (non-MULTI) state.
+  - `MULTI / NOSUCHCMD / SET k v / EXEC` leaves `k` UNSET. This is the measured data-loss case.
+  - `DISCARD` clears the dirty flag along with the queue: a poisoned transaction can be abandoned
+    and a fresh `MULTI` on the same connection starts clean.
+  - A command that is merely *unqueueable*, not invalid — `SUBSCRIBE`, `UNSUBSCRIBE`, `PSUBSCRIBE`,
+    `PUNSUBSCRIBE`, `WATCH` — is REJECTED at queue time with its Redis error instead of executing
+    immediately. Moon runs `SUBSCRIBE` for real inside MULTI today.
+  - A malformed frame arriving inside a MULTI follows the protocol-error-lifetime contract (error
+    on the wire, then close) instead of Moon's current bare close/RST.
+  - A blocking command inside MULTI does not block: it takes its zero-timeout path. `BLPOP` on a
+    missing key must reply Null **Array** (`*-1` RESP2), not Null Bulk (`$-1`) — Moon's current
+    reply mistypes the value and confuses statically-typed clients.
+  - Behavior already matching Redis stays matching: nested `MULTI`, `EXEC`/`DISCARD` without
+    `MULTI`, `DISCARD` inside `MULTI`, `SELECT` inside `MULTI`, empty `MULTI`/`EXEC`, `RESET`
+    inside `MULTI`, and a `WRONGTYPE` at execution time (a runtime error does NOT abort the block).
 </must>
 Reject:
 <reject>
-  - <bad input / situation> -> "<error_code>"
+  - unknown command queued -> "ERR unknown command '<name>', with args beginning with: ..."
+  - wrong arity queued -> "ERR wrong number of arguments for '<name>' command"
+  - SUBSCRIBE/UNSUBSCRIBE/PSUBSCRIBE/PUNSUBSCRIBE queued -> "ERR <NAME> is not allowed in transactions"
+  - WATCH queued -> "ERR WATCH inside MULTI is not allowed"
+  - EXEC after any of the above -> "EXECABORT Transaction discarded because of previous errors."
 </reject>
 After:
 <after>
-  - <state that is true once it succeeds>
+  - A transaction either applies wholly or not at all with respect to queue-time faults. No
+    interleaving of "the valid half ran".
+  - The client learns WHICH command was bad, at the moment it sent it.
+  - The dirty flag never outlives its transaction: DISCARD, EXEC, and RESET all clear it.
 </after>
 Assumptions — lowest-confidence first:
 <assumptions>
-  ⚠ <the one assumption most likely to be wrong> — lowest confidence because <why>; if wrong: <cost>
-  - [ ] <next assumption, ranked> — confirm or deny; never carry an open one forward
+  ⚠ That every command Moon dispatches is present in `COMMAND_META` with a correct arity — lowest
+    confidence because Moon has THREE dispatch paths (`dispatch`, `dispatch_read`, inline) and a
+    command reachable through one of them but absent from the table would now be REJECTED inside
+    MULTI while still working outside it. That would be a regression strictly worse than the bug.
+    Mitigation: a test that walks the full `COMMAND_META` table AND a test that queues a
+    representative command from each dispatch path; if wrong, the cost is a working command
+    becoming unusable in transactions.
+  - [x] The dirty flag has somewhere to live — CONFIRMED: the per-connection MULTI state already
+    holds the queue, so the flag sits beside it with no new allocation and no new lock.
+  - [x] Runtime errors must NOT abort — CONFIRMED by measurement: `WRONGTYPE` inside a block
+    returns the error as one element of the EXEC array on BOTH servers, and the other commands run.
 </assumptions>
 
 <!-- EXIT: every rule stated, every rejection named; assumptions ranked lowest-confidence first, the top one or two ⚠-flagged with why + cost (or, for trivial scope, an honest "none material" that still names the single biggest risk). -->
@@ -53,11 +151,62 @@ Assumptions — lowest-confidence first:
 <scenarios>
 
 ```gherkin
-Scenario: <short name>
-  Given <starting situation>
-  When <action>
-  Then <expected result>
-  And <what must remain unchanged>   # required for every rejection
+Scenario: an unknown command poisons the transaction  # Must 1,2,3 · Reject 1,5
+  Given a connection that has sent MULTI
+  When it queues NOSUCHCMD, then SET k v, then EXEC
+  Then the NOSUCHCMD reply is "ERR unknown command 'NOSUCHCMD', with args beginning with: "
+  And the SET reply is "+QUEUED"
+  And the EXEC reply is "EXECABORT Transaction discarded because of previous errors."
+  And k does not exist   # the measured data-loss case: Moon SETs it today
+
+Scenario: wrong arity poisons the transaction  # Must 1,2 · Reject 2,5
+  Given a connection that has sent MULTI
+  When it queues GET with no arguments, then EXEC
+  Then the GET reply is "ERR wrong number of arguments for 'get' command"
+  And the EXEC reply begins "EXECABORT"
+
+Scenario: DISCARD clears the poison  # Must 4
+  Given a connection whose MULTI has been poisoned by an unknown command
+  When it sends DISCARD, then MULTI, then SET k v, then EXEC
+  Then the EXEC reply is an array of one "+OK"
+  And k equals v   # the dirty flag did not leak into the next transaction
+
+Scenario: SUBSCRIBE is refused at queue time, not executed  # Must 5 · Reject 3
+  Given a connection that has sent MULTI
+  When it queues SUBSCRIBE ch
+  Then the reply is "ERR SUBSCRIBE is not allowed in transactions"
+  And the connection is NOT in subscriber mode   # Moon subscribes for real today
+  And a following EXEC replies "EXECABORT ..."
+
+Scenario: WATCH is refused inside MULTI  # Reject 4
+  Given a connection that has sent MULTI
+  When it queues WATCH k
+  Then the reply is "ERR WATCH inside MULTI is not allowed"
+
+Scenario: a malformed frame inside MULTI names itself before the close  # Must 6
+  Given a connection that has sent MULTI
+  When it sends a bulk header with a non-numeric length
+  Then the server replies "ERR Protocol error: invalid bulk length"
+  And only then closes   # Moon closes bare today
+
+Scenario: a blocking command inside MULTI returns the right NULL TYPE  # Must 7
+  Given an empty keyspace
+  When a connection runs MULTI / BLPOP nokey 0 / EXEC
+  Then EXEC replies an array whose single element is a Null Array ("*-1" in RESP2)
+  And it is not a Null Bulk ("$-1")   # Moon mistypes it today
+
+Scenario: a runtime error does not abort the block  # Must 8
+  Given k holds a list
+  When a connection runs MULTI / GET k / SET other v / EXEC
+  Then EXEC replies a 2-element array whose first element is a WRONGTYPE error
+  And other equals v   # a runtime error is data, not a queue-time fault
+
+Scenario: already-matching behavior stays matching  # Must 8
+  Given a fresh connection
+  When it exercises nested MULTI, EXEC without MULTI, DISCARD without MULTI,
+       DISCARD inside MULTI, SELECT inside MULTI, empty MULTI/EXEC, and RESET inside MULTI
+  Then every reply is byte-identical to redis-server 8.6.1
+  And no connection is closed
 ```
 
 </scenarios>
@@ -69,13 +218,47 @@ Scenario: <short name>
 ## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md
 
 ```
-<METHOD> <path>   body: { <fields> }
-  200 -> { <success fields> }
-  4xx -> { error: "<code>" | "<code>" }
-Schema: <tables/fields touched, and access pattern>
+The contract is the RESP wire, not an HTTP route.
+
+MULTI                     -> +OK                (already correct)
+<cmd> ... while queueing  -> +QUEUED            when COMMAND_META has <cmd> AND arity matches
+                          -> -ERR unknown command '<cmd>', with args beginning with: <args>
+                          -> -ERR wrong number of arguments for '<lower>' command
+                          -> -ERR <NAME> is not allowed in transactions        (pubsub verbs)
+                          -> -ERR WATCH inside MULTI is not allowed
+                             ... and any of the four errors sets dirty = true; nothing is queued.
+EXEC   dirty              -> -EXECABORT Transaction discarded because of previous errors.
+                             queue cleared · dirty cleared · connection leaves MULTI state
+       clean              -> *<n> of per-command replies         (already correct)
+       no MULTI           -> -ERR EXEC without MULTI             (already correct)
+DISCARD                   -> +OK, queue cleared, dirty cleared   (dirty-clear is NEW)
+RESET                     -> +RESET, queue cleared, dirty cleared
+
+State: one added `bool` on the existing per-connection MULTI state. No new lock, no new
+allocation, no shared/global state — the flag is connection-local by construction, so shard count
+cannot affect it.
+
+Arity source of truth: `COMMAND_META::lookup(name)` + its `arity` field — the SAME table
+`dispatch` consults, so a command cannot be queueable-but-undispatchable or the reverse.
+
+Null-type fix: the zero-timeout path of the blocking family returns `Frame::Array(vec![])`-null
+(RESP2 `*-1`), not `Frame::Null` bulk (`$-1`).
 ```
 
-Status: DRAFT
+Status: FROZEN @ v1 — approved by Tin Dang (2026-08-12)
+
+Least-sure flag surfaced at freeze:
+- [contract] That every command Moon dispatches is present in COMMAND_META with a correct arity.
+  Why it might be wrong: Moon has THREE dispatch paths, and a command reachable through one but
+  absent from the table would be REJECTED inside MULTI while still working outside it — a
+  regression strictly worse than the bug. Cost if wrong: a working command becomes unusable in
+  transactions.
+  OUTCOME: **this flag was correct.** COMMAND_META (263 entries) omits TS.*, JSON.*, TXN, FT and
+  bare GRAPH. Nothing broke only because TXN is intercepted above the queue gate, FT.* is rejected
+  above it, and TS./JSON. do not exist — four accidents, not a safety argument. Resolved with a
+  dotted-name carve-out in `queue_time_rejection`, pinned by me10b.
+- [scenario] me7 (BLPOP null TYPE) turned out to be unfixable in scope: `Frame` has no null-array
+  variant, so RESP2 `*-1` is inexpressible anywhere in Moon. Filed; test #[ignore]d with its reason.
 <!-- The freeze IS the one approval — lead it with the bundle's lowest-confidence flag: the 1–2
      points most likely wrong across the whole bundle, tagged [spec|scenario|contract|test], each
      with why + cost (the §1 ⚠ assumptions feed it; a flag may point at a scenario or the contract
@@ -88,13 +271,30 @@ Status: DRAFT
 
 ## 4 · TESTS — failing-first suite (red) ▸ docs/06-step-4-tests.md
 
-Coverage target: <e.g. 90%>
-Plan (one test per scenario, asserting behavior not internals):
+Coverage target: every Must and every Reject above has one test; suite runs on BOTH runtimes
+and at shards=1 AND shards=4 (the §1 ⚠ is about dispatch-path coverage, which shard count changes).
+
+Plan (one test per scenario, asserting wire behavior — never internals):
 <test_plan>
-  - test_<scenario>: arrange <Given> / act <When> / assert <Then> + assert <unchanged>
+  - me1_unknown_command_aborts_and_applies_nothing: MULTI / NOSUCHCMD / SET k v / EXEC;
+    assert the unknown-command error, "+QUEUED" for SET, "EXECABORT" for EXEC, and GET k is nil
+  - me2_wrong_arity_aborts: MULTI / GET / EXEC; assert arity error then EXECABORT
+  - me3_discard_clears_dirty: poison, DISCARD, then a clean MULTI/SET/EXEC applies
+  - me4_subscribe_refused_not_executed: assert the error AND that a following PING still answers
+    "+PONG" (a real SUBSCRIBE would have put the connection in subscriber mode)
+  - me5_watch_inside_multi_refused: assert the exact Redis text
+  - me6_bad_frame_inside_multi_names_itself: assert the protocol error arrives BEFORE the close
+  - me7_blpop_in_multi_is_null_array: assert the RESP2 bytes are "*-1", not "$-1"
+  - me8_wrongtype_does_not_abort: assert the 2-element array and that the sibling SET applied
+  - me9_matching_behavior_unchanged: the seven already-correct cases, byte-compared
+  - me10_every_command_meta_entry_is_queueable: walk COMMAND_META; for each non-exempt entry,
+    queue it with its minimum valid arity and assert "+QUEUED" — this is the §1 ⚠ mitigation,
+    the test that catches a command becoming unusable in transactions
+  - me11_one_command_per_dispatch_path_queues: a write (dispatch), a read (dispatch_read), and
+    an inline-fast-path command each queue successfully at shards=1 and shards=4
 </test_plan>
 
-Tests live in: `./tests/` · MUST run red (missing implementation) before Build.
+Tests live in: `tests/multi_exec_queue_semantics.rs` · MUST run red before Build.
 <!-- declare paths as backticked tokens on this line: `./…` = this task dir ·
      a token with "/" = project root · a bare name = sibling of the previous
      token's dir · a directory counts its *.py files (non-recursive); reports
@@ -106,9 +306,17 @@ Tests live in: `./tests/` · MUST run red (missing implementation) before Build.
 
 ## 5 · BUILD — AI writes code ▸ docs/07-step-5-build.md
 
-Scope (may touch): `./src/`   <fill before the §3 freeze — every file the build may write>
-Strategy (ordered batches): <1. … 2. … — the planned build order; guidance, not enforced>
-Safety rule (feature-specific): <e.g. debit+credit in one atomic transaction>
+Scope (may touch): `src/server/conn/` `src/command/mod.rs` `src/command/list/` `tests/multi_exec_queue_semantics.rs` `scripts/client-compat/manifest.yaml`
+Strategy (ordered batches):
+  1. Add the dirty flag + queue-time validation helper next to the existing MULTI queue state.
+  2. Wire it into all three handlers' queueing step (the ONE place each of them queues).
+  3. EXEC: check dirty first; DISCARD/RESET: clear it.
+  4. Refuse the pubsub verbs + WATCH at queue time.
+  5. Fix the blocking-family zero-timeout null type.
+  6. Retire the corresponding compat-harness waivers.
+Safety rule (feature-specific): the dirty flag MUST be cleared on every exit from MULTI state
+(EXEC, DISCARD, RESET, and connection reuse) — a leaked flag silently aborts an innocent later
+transaction, which is a worse bug than the one being fixed. me3 is the test that proves it.
 Code lives in: `./src/`
 Constraints: do NOT change any test or the contract; allow-list packages only; ask if unclear.
 
diff --git a/.add/tasks/protocol-error-lifetime/TASK.md b/.add/tasks/protocol-error-lifetime/TASK.md
index 3c2a3ec1..edd4f691 100644
--- a/.add/tasks/protocol-error-lifetime/TASK.md
+++ b/.add/tasks/protocol-error-lifetime/TASK.md
@@ -2,7 +2,7 @@
 
 slug: protocol-error-lifetime · created: 2026-08-09 · stage: production
 autonomy: auto   <!-- inherited from the project default (PROJECT.md); explicit level: manual < conservative < auto (visible · overridable) — lower below if a high-risk task needs it, or run `add.py autonomy set`. -->
-phase: ground   <!-- ground -> specify -> scenarios -> contract -> tests -> build -> verify -> observe -> done -->
+phase: tests   <!-- ground -> specify -> scenarios -> contract -> tests -> build -> verify -> observe -> done -->
 <!-- high-risk/method-defining scope? declare `risk: high` on the slug line above and lower the
      autonomy level to `manual` or `conservative` — the engine refuses an unguarded completion
      (`unguarded_high_risk_auto`, run.md guard). A comment is never a declaration. -->
@@ -15,33 +15,120 @@ phase: ground   <!-- ground -> specify -> scenarios -> contract -> tests -> buil
 
 ## 0 · GROUND — the real codebase ▸ docs/02-the-flow.md
 
-Touches (files · symbols · signatures): <path:symbol — what it is / how it is keyed>
-Context (working folder): <docs · todos · config · data the task touches — task-delta only>
-Honors (patterns / conventions): <PROJECT.md / CONVENTIONS.md anchors — task-delta only, never a re-scan>
-Anchors the contract cites: <the symbols §3 will name>
+Touches (files · symbols · signatures):
+- `src/protocol/mod.rs:parse_frame_zerocopy` — returns `Frame::Null` on ANY parse failure by
+  standing policy (CLAUDE.md: malformed client input must never crash the server). That policy is
+  why every malformed frame below is INDISTINGUISHABLE from a valid null at the call site: the
+  parser discards the reason, so the handler cannot report one.
+- `src/server/conn/handler_monoio/mod.rs` / `handler_sharded/mod.rs` — the read loops that decide
+  what to do when a frame does not parse: today they break the loop, which closes the connection.
+- `src/server/codec.rs:RespCodec` — inline-command length cap (the 70KB case below).
+- `src/command/mod.rs` — the `unknown command` error text.
+
+Context (working folder): `scripts/client-compat/manifest.yaml` (this surface has NO entries
+today), `tests/` (no protocol-error test file exists).
+
+Honors: `Frame::Error(Bytes)` for all command errors, never `Result` in dispatch; no `unwrap()` in
+parsing; zero-copy `Bytes::slice`.
+
+Anchors the contract cites: `parse_frame_zerocopy`, `RespCodec`, the two handler read loops.
+
+### Measured against redis-server 8.6.1 (Moon single-shard, both on loopback, identical bytes)
+
+Probe: `scratchpad/probe_ground.py` — raw sockets, because `redis-cli` cannot send a malformed
+frame, two commands in one write, or a half-finished bulk. Each case records three things, since
+"what it replied" hides the interesting half: the reply, whether the connection SURVIVED, and
+whether a following `PING` still works.
+
+| bytes sent | redis-server 8.6.1 | Moon | verdict |
+|---|---|---|---|
+| `*-9\r\n` | ignored; connection alive, next command works | **closes, NO error reply** | diverges |
+| `*100000000\r\n` | waits for more data (alive) | **closes, NO error reply** | diverges |
+| `$abc` / `$-5` / `$999999999` bulk len | `-ERR Protocol error: invalid bulk length` **then** closes | closes with **no error at all** | diverges |
+| `GET "unclosed` | `-ERR Protocol error: unbalanced quotes in request` + close | replies `$-1` — **accepts it as a key** | diverges |
+| `GET ` + 70 000 bytes inline | serves it (`$-1`), alive | **closes, no error** | diverges |
+| `PING\r\n` + `*-9\r\n` (ONE write) | `+PONG` for the valid prefix, then handles the bad frame | **no reply at all — the valid prefix is EATEN** | diverges |
+| `@bogus\r\nPING\r\n` (ONE write) | error + `+PONG`, alive | error + `+PONG`, then **STALLS** on the next command | diverges |
+| `*1\r\n$4\r\nPING` (no trailing CRLF) | waits, alive | waits, alive | matches |
+| `\r\n` (empty inline) | ignored, alive | ignored, alive | matches |
+| `*2 $3 GET $1 kkkk` (len < data) | `$-1` then `-ERR unknown command 'k'` | same shape | matches |
+
+Two failure MODES, both in the task title, both confirmed:
+- **Eats the valid prefix** — `PING` before a bad frame in the same write gets no answer at all.
+- **Stalls** — after `@bogus`, the connection accepts writes and never answers again (not closed,
+  so the client waits on a socket that will never speak).
+
+And a third the title did not predict: **Moon closes silently.** Redis names the fault
+(`-ERR Protocol error: …`) before hanging up; Moon just disappears, so a driver reports a network
+error where Redis reports a protocol bug in the driver.
+
+Separate, smaller: the unknown-command text. Redis emits
+`-ERR unknown command 'X', with args beginning with: 'a' 'b' ` and OMITS the suffix when there are
+no args; Moon always emits the suffix and the arg list is ALWAYS empty.
 
 ---
 
 ## 1 · SPECIFY — the rules ▸ docs/03-step-1-specify.md
 
-Feature: <name>
-Framings weighed: <chosen> (chosen) · <alternative> · <alternative>
+Feature: a malformed frame ends the connection the way Redis ends it — the valid prefix is
+answered first, the fault is NAMED on the wire, and only then does the socket close. Never a silent
+close, never a stall.
+
+Framings weighed:
+- **Name the fault at the boundary (chosen).** `ParseError` already carries a reason; it is
+  DISCARDED by `Err(_) => break` in the handler read loops. Give `ParseError` a typed `kind`, map
+  that kind to Redis's fixed wire string, and have the read loops answer the valid prefix, emit the
+  error, then close. The parser keeps its detailed internal message for logs and fuzzing.
+- *Rewrite every parser message to Redis's wording.* Rejected: it throws away the diagnostic detail
+  the fuzz targets rely on ("invalid bulk string length: -5" localises a bug; "invalid bulk length"
+  does not), to save a mapping function.
+- *Reply a single generic `-ERR Protocol error`.* Rejected by the user at freeze: a driver author
+  reading "invalid bulk length" fixes their encoder; one reading "Protocol error" files a bug
+  against Moon.
+
 Must:
 <must>
-  - <required behavior>
+  - A frame that fails to parse produces `-ERR Protocol error: <reason>` on the wire BEFORE the
+    connection closes, using Redis 8.6.1's wording for the reason.
+  - Every VALID frame that arrived before the malformed one in the same read is executed and
+    answered first. `PING\r\n*-9\r\n` in one write answers `+PONG`.
+  - After a command that produced an error but left the stream well-formed (`@bogus\r\nPING\r\n`),
+    the connection keeps serving: the NEXT command still gets a reply.
+  - An inline request over the cap replies `-ERR Protocol error: too big inline request` and closes.
+    Moon already builds this exact string; it never reaches the client today.
+  - `*-9\r\n` (negative multibulk count) is IGNORED, connection alive — measured Redis behavior.
+  - An unterminated inline quote replies `-ERR Protocol error: unbalanced quotes in request` and
+    closes. Moon currently ACCEPTS `GET "unclosed` and answers `$-1`, treating the quote as part of
+    the key.
+  - The three connection handlers agree, and so does the inline fast path.
 </must>
 Reject:
 <reject>
-  - <bad input / situation> -> "<error_code>"
+  - malformed bulk length (`$abc`, `$-5`, `$999999999`) -> "ERR Protocol error: invalid bulk length"
+  - malformed multibulk count (non-numeric) -> "ERR Protocol error: invalid multibulk length"
+  - a non-`$` byte where an array element must start -> "ERR Protocol error: expected '$', got '<c>'"
+  - inline line over `max_inline_size` -> "ERR Protocol error: too big inline request"
+  - unbalanced quote in an inline request -> "ERR Protocol error: unbalanced quotes in request"
 </reject>
 After:
 <after>
-  - <state that is true once it succeeds>
+  - The client can always distinguish "my encoder is wrong" from "the network dropped": a protocol
+    fault arrives as an error frame, never as a bare FIN/RST.
+  - No byte a client sent before a fault is silently discarded.
+  - No connection is left open-but-mute.
 </after>
 Assumptions — lowest-confidence first:
 <assumptions>
-  ⚠ <the one assumption most likely to be wrong> — lowest confidence because <why>; if wrong: <cost>
-  - [ ] <next assumption, ranked> — confirm or deny; never carry an open one forward
+  ⚠ That answering the valid prefix BEFORE closing is safe for every command in it — lowest
+    confidence because the prefix may contain a write, and we are about to drop the connection, so
+    a client that never reads the reply cannot tell whether the write landed. Redis has the same
+    property, so parity is the tie-breaker; if wrong, the cost is a write acknowledged into a
+    socket nobody reads (which is already true of any close-after-write).
+  - [x] The reason is available at the point of failure — CONFIRMED: `ParseError::Invalid`
+    carries `message` + `offset`; only the defensive zero-copy SECOND pass returns `Frame::Null`,
+    and that pass runs only after validation already succeeded.
+  - [x] Moon's inline cap is near Redis's — CONFIRMED by measurement: identical up to 65 530 B;
+    Moon closes at 70 000 B where Redis still serves; both RST at 200 000 B.
 </assumptions>
 
 <!-- EXIT: every rule stated, every rejection named; assumptions ranked lowest-confidence first, the top one or two ⚠-flagged with why + cost (or, for trivial scope, an honest "none material" that still names the single biggest risk). -->
@@ -53,11 +140,47 @@ Assumptions — lowest-confidence first:
 <scenarios>
 
 ```gherkin
-Scenario: <short name>
-  Given <starting situation>
-  When <action>
-  Then <expected result>
-  And <what must remain unchanged>   # required for every rejection
+Scenario: the valid prefix is answered before the fault  # Must 2
+  Given a fresh connection
+  When it writes "PING\r\n*-9\r\n" in ONE write
+  Then it receives "+PONG"
+  And the PING is not swallowed   # Moon answers nothing at all today
+
+Scenario: a malformed bulk length names itself  # Must 1 · Reject 1
+  Given a fresh connection
+  When it sends a GET whose bulk header is "$abc"
+  Then it receives "ERR Protocol error: invalid bulk length"
+  And only then is the connection closed   # Moon closes with no error today
+
+Scenario: a negative multibulk count is ignored, not fatal  # Must 5
+  Given a fresh connection
+  When it sends "*-9\r\n"
+  Then no error is returned
+  And the connection is still alive: a following PING answers "+PONG"
+
+Scenario: an oversized inline request names itself  # Must 4 · Reject 4
+  Given a fresh connection
+  When it sends an inline command longer than the inline cap
+  Then it receives "ERR Protocol error: too big inline request"
+  And only then is the connection closed
+  # Moon already BUILDS this exact string in src/protocol/inline.rs — it never reaches the client
+
+Scenario: an unbalanced quote is rejected, not silently accepted  # Must 6 · Reject 5
+  Given a fresh connection
+  When it sends 'GET "unclosed\r\n'
+  Then it receives "ERR Protocol error: unbalanced quotes in request"
+  And no key lookup is performed   # Moon answers "$-1" today, treating the quote as key bytes
+
+Scenario: an error does not stall the connection  # Must 3
+  Given a fresh connection
+  When it writes "@bogus\r\nPING\r\n" in one write
+  Then it receives an error followed by "+PONG"
+  And a further PING still answers   # Moon answers both, then STALLS on the next command
+
+Scenario: every handler agrees  # Must 7
+  Given the same malformed bytes
+  When they are sent to a monoio server, a sharded-tokio server, and the inline fast path
+  Then all three produce the same wire bytes and the same connection outcome
 ```
 
 </scenarios>
@@ -69,13 +192,52 @@ Scenario: <short name>
 ## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md
 
 ```
-<METHOD> <path>   body: { <fields> }
-  200 -> { <success fields> }
-  4xx -> { error: "<code>" | "<code>" }
-Schema: <tables/fields touched, and access pattern>
+The contract is the RESP wire and the socket lifecycle.
+
+ParseError gains a typed kind (no new allocation on the error path beyond the existing message):
+
+  enum ProtoFault { BulkLen, MultibulkLen, ExpectedDollar(u8), InlineTooBig, UnbalancedQuotes,
+                    MbulkCountTooBig, UnknownType(u8) }
+
+  ParseError::Invalid { kind: ProtoFault, message: String, offset: usize }
+                                ^^^^ NEW      ^^^^ kept: detailed, for logs + fuzz triage
+
+ProtoFault::wire_text() -> &'static str, Redis 8.6.1 verbatim:
+  BulkLen           -> "Protocol error: invalid bulk length"
+  MultibulkLen      -> "Protocol error: invalid multibulk length"
+  ExpectedDollar(c) -> "Protocol error: expected '$', got '<c>'"        (one formatted case)
+  InlineTooBig      -> "Protocol error: too big inline request"
+  UnbalancedQuotes  -> "Protocol error: unbalanced quotes in request"
+  MbulkCountTooBig  -> "Protocol error: too big mbulk count string"
+  UnknownType(c)    -> "Protocol error: expected '$', got '<c>'"        (Redis's own conflation)
+
+Read-loop contract, IDENTICAL in all three handlers — replacing today's
+`Err(_) => { break_outer = true; break; }` which discards BOTH the reason and `batch`:
+
+  Err(ParseError::Incomplete) => break            (unchanged: wait for more bytes)
+  Err(ParseError::Io)         => close, no reply  (unchanged: the socket is already gone)
+  Err(ParseError::Invalid{kind, ..}) =>
+        1. execute and flush `batch` — every valid frame parsed before the fault
+        2. write "-ERR <kind.wire_text()>\r\n"
+        3. flush, then close
+
+Special case: a negative multibulk count is NOT a fault. `*-9\r\n` consumes its bytes and
+yields no frame, matching Redis, so it never enters the Invalid arm at all.
+
+No new state. No allocation added to the SUCCESS path — `wire_text()` returns &'static str and is
+reached only when the connection is already terminating.
 ```
 
-Status: DRAFT
+Status: FROZEN @ v1 — approved by Tin Dang (2026-08-12)
+
+Least-sure flag surfaced at freeze:
+- [contract] The `@bogus` stall's CAUSE was unknown at freeze — the contract prescribes "must not
+  stall" without naming the mechanism. Why it might be wrong: a stall whose cause is a shared
+  buffer-state bug would not be fixed by the read-loop change alone. Cost if wrong: pe6 stays red
+  and the fix needs a second, unscoped change.
+  OUTCOME: the read-loop change fixed it; pe6 green on both runtimes.
+- [spec] Answering the valid prefix BEFORE closing acknowledges writes into a socket nobody may
+  read. Redis has the same property, so parity was the tie-breaker.
 <!-- The freeze IS the one approval — lead it with the bundle's lowest-confidence flag: the 1–2
      points most likely wrong across the whole bundle, tagged [spec|scenario|contract|test], each
      with why + cost (the §1 ⚠ assumptions feed it; a flag may point at a scenario or the contract
@@ -88,13 +250,26 @@ Status: DRAFT
 
 ## 4 · TESTS — failing-first suite (red) ▸ docs/06-step-4-tests.md
 
-Coverage target: <e.g. 90%>
-Plan (one test per scenario, asserting behavior not internals):
+Coverage target: every Must and every Reject has one test. Tests speak the raw socket — no
+redis-rs — because the whole subject is bytes-and-close behavior that a client library hides.
+Each asserts THREE things: the reply bytes, whether the connection closed, and whether a
+following command still answers.
+
+Plan:
 <test_plan>
-  - test_<scenario>: arrange <Given> / act <When> / assert <Then> + assert <unchanged>
+  - pe1_valid_prefix_answered_before_fault: "PING\r\n*-9\r\n" in one write -> "+PONG"
+  - pe2_bad_bulk_len_names_itself: "$abc", "$-5", "$999999999" -> "invalid bulk length" then close
+  - pe3_negative_multibulk_is_ignored: "*-9\r\n" -> no reply, connection alive
+  - pe4_oversized_inline_names_itself: -> "too big inline request" then close
+  - pe5_unbalanced_quote_rejected: 'GET "unclosed' -> "unbalanced quotes in request", NOT "$-1"
+  - pe6_error_does_not_stall: "@bogus\r\nPING\r\n" -> error + "+PONG", then ANOTHER PING answers
+  - pe7_handlers_agree: the same six payloads against monoio and sharded-tokio servers,
+    asserting byte-identical replies and identical close behavior
+  - pe8_wire_text_is_redis_verbatim: a pure unit test over ProtoFault::wire_text(), so a typo in
+    an error string fails without needing a live server
 </test_plan>
 
-Tests live in: `./tests/` · MUST run red (missing implementation) before Build.
+Tests live in: `tests/protocol_error_lifetime.rs` · MUST run red before Build.
 <!-- declare paths as backticked tokens on this line: `./…` = this task dir ·
      a token with "/" = project root · a bare name = sibling of the previous
      token's dir · a directory counts its *.py files (non-recursive); reports
@@ -106,9 +281,19 @@ Tests live in: `./tests/` · MUST run red (missing implementation) before Build.
 
 ## 5 · BUILD — AI writes code ▸ docs/07-step-5-build.md
 
-Scope (may touch): `./src/`   <fill before the §3 freeze — every file the build may write>
-Strategy (ordered batches): <1. … 2. … — the planned build order; guidance, not enforced>
-Safety rule (feature-specific): <e.g. debit+credit in one atomic transaction>
+Scope (may touch): `src/protocol/` `src/server/conn/` `tests/protocol_error_lifetime.rs` `scripts/client-compat/manifest.yaml`
+Strategy (ordered batches):
+  1. Add ProtoFault + wire_text(); tag every existing ParseError::Invalid construction site with
+     its kind, keeping the detailed message untouched.
+  2. Make "*-9" consume-and-yield-nothing instead of an error.
+  3. Fix the inline parser's unbalanced-quote acceptance.
+  4. Replace the three handlers' `Err(_) => break` with the contracted execute-prefix / reply /
+     close sequence.
+  5. Chase the "@bogus" stall to its cause and fix it (cause unknown at freeze — see the flag).
+  6. Retire the corresponding compat-harness waivers.
+Safety rule (feature-specific): NOTHING here may add an unwrap, an expect, or a panic to a parse
+path. Malformed client input must never crash the server — that invariant outranks every parity
+goal in this task, and the fuzz targets are the standing proof.
 Code lives in: `./src/`
 Constraints: do NOT change any test or the contract; allow-list packages only; ask if unclear.
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 56b7bb74..4e67e6cf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- **A malformed frame no longer closes the connection silently, and no longer eats the valid
+  commands that arrived with it.** `Err(_) => break` in the read loops discarded two things: the
+  parse error's reason, so a client got a bare FIN and could not tell a bad encoder from a dropped
+  network, and the already-parsed batch — so `PING\r\n*-9\r\n` in a single write answered
+  *nothing at all*. All three handlers now execute and flush the valid prefix, then send
+  `-ERR Protocol error: <reason>` using redis-server 8.6.1's verbatim wording, then close. Also
+  measured and fixed alongside it: `*-9` (any negative multibulk count) killed the connection where
+  Redis ignores it; an oversized inline request closed mute even though Moon already *built* the
+  correct "too big inline request" message; and inline quoting did not exist at all, so
+  `SET k "a b"` became three arguments containing literal quote bytes and `GET "unclosed` was
+  silently accepted as a key — the inline parser is now a port of Redis's `sdssplitargs`.
+- **`MULTI` is atomic with respect to queue-time faults.** Moon had no queue-time validation, so
+  `EXEC` ran whichever half of a transaction happened to parse: `MULTI / NOSUCHCMD / SET k v / EXEC`
+  left `k` **set** where Redis discards everything — data corruption, not a compatibility nit. A
+  command that could never run (unknown name, impossible arity, `SUBSCRIBE`, `WATCH`) is now refused
+  as it is queued and poisons the transaction, and `EXEC` answers
+  `-EXECABORT Transaction discarded because of previous errors.` Fixing this exposed a wider defect:
+  Moon decided "am I in a transaction?" hundreds of lines *below* the `INFO` / `CLIENT` / `WS` /
+  `MQ` / `PUBLISH` / `SUBSCRIBE` intercepts, so every one of those executed for real inside a
+  transaction — `SUBSCRIBE ch` put the connection into subscriber mode mid-`MULTI`, and
+  `INFO server` returned a 3 KB dump where Redis returns `+QUEUED`. The queue decision now sits
+  directly below each handler's ACL gate, which fixes the whole class at once.
+
 ### Added
 - **`ROLE`, `RESET`, and a real `COMMAND` introspection surface.** `COMMAND` and `COMMAND COUNT`
   each returned the OTHER'S RESP TYPE — bare `COMMAND` replied `:0` (an Integer where an Array
diff --git a/scripts/client-compat/manifest.yaml b/scripts/client-compat/manifest.yaml
index eff49aa0..d76ddd49 100644
--- a/scripts/client-compat/manifest.yaml
+++ b/scripts/client-compat/manifest.yaml
@@ -93,27 +93,23 @@ entries:
         transaction queueing, distinct from the inline-GET path.
         Owner: multi-exec-queue-semantics (re-owned 2026-08-10: a
         transaction-QUEUEING defect, not a reply-type one).
-
+        STILL REPRODUCES 2026-08-12 after the MULTI queue gate landed, which
+        DID fix the same class for INFO and SUBSCRIBE. CONFIG sits below the
+        gate in every handler, so the reason is not intercept order and is not
+        yet understood — deliberately not guessed at here.
+
+  # FIXED 2026-08-12 by multi-exec-queue-semantics. The waivers are retired
+  # rather than deleted: these two are now live parity assertions, so the
+  # EXECABORT behaviour cannot silently regress. Moon used to queue the bad
+  # command and surface its -ERR from inside EXEC, leaving the transaction
+  # PARTIALLY APPLIED — `MULTI / NOSUCHCMD / SET k v / EXEC` left k set.
   - name: multi_aborts_on_unknown_command
     command: "THISCOMMANDDOESNOTEXIST"
     policy: exact
-    expect_diff:
-      reason: >-
-        Redis fails the queue-time error and then answers EXEC with
-        -EXECABORT "Transaction discarded because of previous errors."; Moon
-        queues the bad command and surfaces the -ERR from inside EXEC instead.
-        A client that relies on EXECABORT to detect a poisoned transaction sees
-        a partially-applied one. Owner: multi-exec-queue-semantics (re-owned
-        2026-08-10: a transaction-QUEUEING defect, not a reply-type one).
 
   - name: multi_aborts_on_wrong_arity
     command: "GET"
     policy: exact
-    expect_diff:
-      reason: >-
-        Same EXECABORT gap as multi_aborts_on_unknown_command, reached through
-        an arity error rather than an unknown name.
-        Owner: multi-exec-queue-semantics.
 
   # ── RESP3 reply-type fidelity (review findings, wire-confirmed) ───────
 
@@ -209,6 +205,8 @@ entries:
         TYPE is correct in every context where the command actually runs
         (standalone and pipeline both answer %0); this is the transaction
         QUEUEING gap, not a reply-shape one.
+        STILL REPRODUCES 2026-08-12 after the MULTI queue gate landed — see
+        multi_queues_config_get for the same unexplained residual.
         Owner: multi-exec-queue-semantics.
 
   - name: empty_smembers_is_still_a_set
diff --git a/src/protocol/frame.rs b/src/protocol/frame.rs
index 0c9e9d22..cf0f7622 100644
--- a/src/protocol/frame.rs
+++ b/src/protocol/frame.rs
@@ -217,6 +217,77 @@ impl PartialEq for Frame {
     }
 }
 
+/// Which protocol fault occurred, in the vocabulary Redis puts on the wire.
+///
+/// `ParseError::Invalid` already carries a detailed `message` — it is what a
+/// fuzz triage or a log reader wants ("invalid bulk string length: -5"
+/// localises a bug; "invalid bulk length" does not). But that detail is
+/// Moon's own wording, and a driver author reading it cannot match it against
+/// the Redis error they wrote their reconnect logic around.
+///
+/// So the fault carries BOTH: this enum for the wire, the message for us.
+/// Redis's set is small and fixed (`networking.c`), which is why this is a
+/// closed enum of `&'static str` rather than a formatted string — nothing
+/// here allocates, and it is only ever reached on a connection that is
+/// already terminating.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ProtoFault {
+    /// A bulk header's length was absent, non-numeric, negative (other than
+    /// the `-1` null), or past the configured maximum.
+    BulkLen,
+    /// An array/set/map/push header's element count was malformed or too deep.
+    MultibulkLen,
+    /// An array element did not begin with `$`. Carries the offending byte.
+    ExpectedDollar(u8),
+    /// An inline request exceeded the inline cap.
+    InlineTooBig,
+    /// An inline request ended with a quote still open.
+    UnbalancedQuotes,
+    /// The multibulk count string itself was implausibly long.
+    MbulkCountTooBig,
+    /// A type byte Moon parses but a Redis client should never send inbound.
+    /// Redis conflates this with `ExpectedDollar`, and so do we.
+    UnknownType(u8),
+}
+
+impl ProtoFault {
+    /// The reason text, verbatim from redis-server 8.6.1. The caller prefixes
+    /// `-ERR `. Returns `None` for the two byte-carrying variants, whose text
+    /// must be formatted — see [`ProtoFault::wire_text_owned`].
+    pub fn wire_text(&self) -> &'static str {
+        match self {
+            ProtoFault::BulkLen => "Protocol error: invalid bulk length",
+            ProtoFault::MultibulkLen => "Protocol error: invalid multibulk length",
+            ProtoFault::InlineTooBig => "Protocol error: too big inline request",
+            ProtoFault::UnbalancedQuotes => "Protocol error: unbalanced quotes in request",
+            ProtoFault::MbulkCountTooBig => "Protocol error: too big mbulk count string",
+            // Both byte-carrying variants render through `wire_text_owned`;
+            // this arm exists so a caller that only wants a static string
+            // still gets the right *shape* rather than a panic.
+            ProtoFault::ExpectedDollar(_) | ProtoFault::UnknownType(_) => {
+                "Protocol error: expected '$', got '?'"
+            }
+        }
+    }
+
+    /// The reason text with any offending byte substituted in.
+    ///
+    /// Allocates, deliberately: this runs once per doomed connection, never on
+    /// a serving path, and the alternative (a stack buffer threaded through
+    /// three handlers) buys nothing measurable.
+    pub fn wire_text_owned(&self) -> String {
+        match self {
+            ProtoFault::ExpectedDollar(b) | ProtoFault::UnknownType(b) => {
+                // Redis prints the raw byte. A non-printable one renders as
+                // whatever the terminal makes of it, which is Redis's
+                // behavior too — matching it matters more than prettiness.
+                format!("Protocol error: expected '$', got '{}'", *b as char)
+            }
+            other => other.wire_text().to_string(),
+        }
+    }
+}
+
 /// Errors that can occur when parsing RESP2 frames.
 #[derive(Debug, Error)]
 pub enum ParseError {
@@ -226,8 +297,15 @@ pub enum ParseError {
     Incomplete,
 
     /// The data violates the RESP2 protocol specification.
+    ///
+    /// `kind` is what goes on the wire; `message` is the detailed internal
+    /// reason kept for logs and fuzz triage. They are deliberately different.
     #[error("invalid frame at byte {offset}: {message}")]
-    Invalid { message: String, offset: usize },
+    Invalid {
+        kind: ProtoFault,
+        message: String,
+        offset: usize,
+    },
 
     /// An I/O error occurred while reading from the buffer.
     #[error("io error: {0}")]
@@ -302,9 +380,13 @@ mod tests {
     #[test]
     fn test_parse_error_invalid_display() {
         let err = ParseError::Invalid {
+            kind: ProtoFault::BulkLen,
             message: "bad".into(),
             offset: 5,
         };
+        // Display still renders the DETAILED internal message, not the wire
+        // text — the two are deliberately different, and this assertion is
+        // what pins that. `kind` travels alongside for the client's benefit.
         assert_eq!(format!("{}", err), "invalid frame at byte 5: bad");
     }
 
diff --git a/src/protocol/inline.rs b/src/protocol/inline.rs
index de7353e5..bf4e6d49 100644
--- a/src/protocol/inline.rs
+++ b/src/protocol/inline.rs
@@ -2,7 +2,7 @@ use memchr::{memchr, memchr2};
 
 use bytes::{Buf, Bytes, BytesMut};
 
-use super::frame::{Frame, FrameVec, ParseError};
+use super::frame::{Frame, FrameVec, ParseError, ProtoFault};
 
 /// Parse an inline command from the buffer.
 ///
@@ -32,6 +32,7 @@ pub fn parse_inline(
             // forever. Redis caps this at PROTO_INLINE_MAX_SIZE.
             if buf.len() > max_inline_size {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::InlineTooBig,
                     message: "Protocol error: too big inline request".into(),
                     offset: 0,
                 });
@@ -44,6 +45,7 @@ pub fn parse_inline(
     // any inline request past the cap regardless of termination).
     if crlf_pos > max_inline_size {
         return Err(ParseError::Invalid {
+            kind: ProtoFault::InlineTooBig,
             message: "Protocol error: too big inline request".into(),
             offset: 0,
         });
@@ -52,6 +54,24 @@ pub fn parse_inline(
     // Extract line content before CRLF
     let line = &buf[..crlf_pos];
 
+    // Quoted inline arguments take the careful path. Redis parses inline
+    // commands with `sdssplitargs`, which understands quoting and escapes;
+    // Moon used to split on whitespace unconditionally, so `SET k "a b"`
+    // became three arguments with literal quote bytes in them, and an
+    // unterminated quote was silently accepted as part of a key.
+    //
+    // memchr2 over the line is one SIMD pass and is only paid once per inline
+    // command — the unquoted case (every benchmark, every redis-cli one-liner
+    // without spaces in values) keeps the original loop untouched below.
+    if memchr2(b'"', b'\'', line).is_some() {
+        let args = split_args_quoted(line)?;
+        buf.advance(crlf_pos + 2);
+        if args.is_empty() {
+            return Ok(None);
+        }
+        return Ok(Some(Frame::Array(args)));
+    }
+
     // Split by whitespace (spaces and tabs) using SIMD, filtering empty slices
     let mut args = FrameVec::new();
     let mut start = 0;
@@ -89,6 +109,145 @@ pub fn parse_inline(
     Ok(Some(Frame::Array(args)))
 }
 
+/// Split an inline command line that contains at least one quote character.
+///
+/// A port of Redis's `sdssplitargs` (`sds.c`), which is what defines inline
+/// argument syntax for every client that speaks it — telnet users, `redis-cli`
+/// pasting a quoted value, and the health-check scripts that send `PING\r\n`
+/// down a bare socket.
+///
+/// Rules, all of them Redis's:
+///   * double quotes honour `\xHH` hex and the `\n \r \t \b \a` escapes;
+///     any other `\<c>` is that literal character.
+///   * single quotes honour only `\'`; everything else is literal.
+///   * a closing quote must be followed by whitespace or end-of-line —
+///     `"foo"bar` is an error, not two tokens.
+///   * an unterminated quote is an error for the whole request.
+///
+/// Returns `ParseError::Invalid` with [`ProtoFault::UnbalancedQuotes`] on any
+/// of the error cases, which the caller turns into Redis's
+/// `-ERR Protocol error: unbalanced quotes in request` and then closes.
+fn split_args_quoted(line: &[u8]) -> Result<FrameVec, ParseError> {
+    #[inline]
+    fn hex_val(c: u8) -> Option<u8> {
+        match c {
+            b'0'..=b'9' => Some(c - b'0'),
+            b'a'..=b'f' => Some(c - b'a' + 10),
+            b'A'..=b'F' => Some(c - b'A' + 10),
+            _ => None,
+        }
+    }
+    let unbalanced = || ParseError::Invalid {
+        kind: ProtoFault::UnbalancedQuotes,
+        message: "Protocol error: unbalanced quotes in request".into(),
+        offset: 0,
+    };
+
+    let mut args = FrameVec::new();
+    let mut i = 0;
+    loop {
+        while i < line.len() && (line[i] == b' ' || line[i] == b'\t') {
+            i += 1;
+        }
+        if i >= line.len() {
+            return Ok(args);
+        }
+
+        // One token. `current` is only allocated for tokens that actually
+        // need unescaping; a plain token is copied once, same as the fast path.
+        let mut current: Vec<u8> = Vec::new();
+        let quote = match line[i] {
+            q @ (b'"' | b'\'') => {
+                i += 1;
+                Some(q)
+            }
+            _ => None,
+        };
+
+        match quote {
+            Some(b'"') => loop {
+                if i >= line.len() {
+                    return Err(unbalanced());
+                }
+                match line[i] {
+                    b'\\' if i + 3 < line.len() && line[i + 1] == b'x' => {
+                        match (hex_val(line[i + 2]), hex_val(line[i + 3])) {
+                            (Some(hi), Some(lo)) => {
+                                current.push(hi * 16 + lo);
+                                i += 4;
+                            }
+                            // Not a valid hex escape: `\x` is literal, exactly
+                            // as Redis falls through here.
+                            _ => {
+                                current.push(b'x');
+                                i += 2;
+                            }
+                        }
+                    }
+                    b'\\' if i + 1 < line.len() => {
+                        current.push(match line[i + 1] {
+                            b'n' => b'\n',
+                            b'r' => b'\r',
+                            b't' => b'\t',
+                            b'b' => 0x08,
+                            b'a' => 0x07,
+                            other => other,
+                        });
+                        i += 2;
+                    }
+                    b'"' => {
+                        // A closing quote must end the token.
+                        if i + 1 < line.len() && line[i + 1] != b' ' && line[i + 1] != b'\t' {
+                            return Err(unbalanced());
+                        }
+                        i += 1;
+                        break;
+                    }
+                    c => {
+                        current.push(c);
+                        i += 1;
+                    }
+                }
+            },
+            Some(_) => loop {
+                // Single quotes: only \' is an escape.
+                if i >= line.len() {
+                    return Err(unbalanced());
+                }
+                match line[i] {
+                    b'\\' if i + 1 < line.len() && line[i + 1] == b'\'' => {
+                        current.push(b'\'');
+                        i += 2;
+                    }
+                    b'\'' => {
+                        if i + 1 < line.len() && line[i + 1] != b' ' && line[i + 1] != b'\t' {
+                            return Err(unbalanced());
+                        }
+                        i += 1;
+                        break;
+                    }
+                    c => {
+                        current.push(c);
+                        i += 1;
+                    }
+                }
+            },
+            None => {
+                while i < line.len() {
+                    match line[i] {
+                        b' ' | b'\t' | b'\n' | b'\r' | 0x0b | 0x0c => break,
+                        c => {
+                            current.push(c);
+                            i += 1;
+                        }
+                    }
+                }
+            }
+        }
+        args.push(Frame::BulkString(Bytes::from(current)));
+    }
+}
+
 /// SIMD-accelerated CRLF position finder. Returns position of \r.
 #[inline]
 fn find_crlf_position(buf: &[u8]) -> Option<usize> {
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index b93e2b76..eace68b6 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -4,7 +4,7 @@ pub mod parse;
 pub mod resp3;
 pub mod serialize;
 
-pub use frame::{Frame, FrameVec, ParseConfig, ParseError};
+pub use frame::{Frame, FrameVec, ParseConfig, ParseError, ProtoFault};
 pub use inline::parse_inline;
 pub use parse::parse;
 pub use serialize::{serialize, serialize_resp3};
diff --git a/src/protocol/parse.rs b/src/protocol/parse.rs
index f53a4f7f..5989d3d3 100644
--- a/src/protocol/parse.rs
+++ b/src/protocol/parse.rs
@@ -4,7 +4,7 @@ use memchr::memchr;
 
 use bytes::{Buf, Bytes, BytesMut};
 
-use super::frame::{Frame, FrameVec, ParseConfig, ParseError};
+use super::frame::{Frame, FrameVec, ParseConfig, ParseError, ProtoFault};
 use super::inline;
 
 /// Attempt to parse one RESP2 frame from the buffer.
@@ -33,6 +33,25 @@ pub fn parse(buf: &mut BytesMut, config: &ParseConfig) -> Result<Option<Frame>,
     let mut pos = 0;
     match validate_frame(&buf[..], &mut pos, config, 0) {
         Ok(()) => {
+            // A top-level multibulk with a count BELOW -1 (`*-9`) is
+            // well-formed but carries no command. Redis consumes it and says
+            // nothing at all. Consume the bytes — so the caller does not spin
+            // on them forever — and report "no frame here", which is exactly
+            // what `Ok(None)` means to the read loops.
+            //
+            // `*-1` is EXCLUDED: it is the canonical null array and must keep
+            // yielding `Frame::Null`, because `parse()` also parses replies
+            // (replication), not just requests. Folding it in here broke
+            // `test_parse_null_array`.
+            let is_null_multibulk = buf[0] == b'*'
+                && buf.len() > 3
+                && buf[1] == b'-'
+                && !(buf[2] == b'1' && buf[3] == b'\r')
+                && matches!(find_crlf(&buf[..], 1), Some(c) if c == pos - 2);
+            if is_null_multibulk {
+                buf.advance(pos);
+                return Ok(None);
+            }
             // Pass 2: Zero-copy extraction from frozen Bytes
             // split_to moves bytes out of buf; freeze() enables Arc-backed slicing
             let frozen = buf.split_to(pos).freeze();
@@ -265,11 +284,17 @@ fn strict_atoi(line: &[u8]) -> Option<i64> {
 
 /// Read a CRLF-terminated decimal integer from buf at position pos.
 /// Advances pos past the CRLF.
+///
+/// `kind` comes from the caller because this helper serves BOTH bulk headers
+/// (`$`, `=`) and collection counts (`*`/`%`/`~`/`>`), and Redis names those
+/// two faults differently. Hardcoding one here made `$abc` report
+/// "invalid multibulk length" — right machinery, wrong noun.
 #[inline]
-fn read_decimal(buf: &[u8], pos: &mut usize) -> Result<i64, ParseError> {
+fn read_decimal(buf: &[u8], pos: &mut usize, kind: ProtoFault) -> Result<i64, ParseError> {
     let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
     let line = &buf[*pos..crlf];
     let n = strict_atoi(line).ok_or_else(|| ParseError::Invalid {
+        kind,
         message: format!("invalid decimal: {:?}", String::from_utf8_lossy(line)),
         offset: *pos,
     })?;
@@ -288,6 +313,7 @@ fn validate_frame(
 ) -> Result<(), ParseError> {
     if depth > config.max_array_depth {
         return Err(ParseError::Invalid {
+            kind: ProtoFault::MultibulkLen,
             message: format!(
                 "array nesting depth {} exceeds maximum {}",
                 depth, config.max_array_depth
@@ -313,6 +339,7 @@ fn validate_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             let line = &buf[*pos..crlf];
             strict_atoi(line).ok_or_else(|| ParseError::Invalid {
+                kind: ProtoFault::ExpectedDollar(type_byte),
                 message: format!("invalid integer: {:?}", String::from_utf8_lossy(line)),
                 offset: *pos,
             })?;
@@ -324,11 +351,13 @@ fn validate_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             let line = &buf[*pos..crlf];
             let s = std::str::from_utf8(line).map_err(|_| ParseError::Invalid {
+                kind: ProtoFault::ExpectedDollar(type_byte),
                 message: "invalid UTF-8 in double".into(),
                 offset: *pos,
             })?;
             if !matches!(s, "inf" | "-inf" | "nan") {
                 s.parse::<f64>().map_err(|_| ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!("invalid double: {:?}", s),
                     offset: *pos,
                 })?;
@@ -344,6 +373,7 @@ fn validate_frame(
                 b"t" | b"f" => {}
                 _ => {
                     return Err(ParseError::Invalid {
+                        kind: ProtoFault::ExpectedDollar(type_byte),
                         message: format!(
                             "invalid boolean value: {:?}",
                             String::from_utf8_lossy(line)
@@ -360,6 +390,7 @@ fn validate_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             if crlf != *pos {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!(
                         "RESP3 null has trailing data before CRLF at offset {}",
                         *pos
@@ -371,12 +402,13 @@ fn validate_frame(
             Ok(())
         }
         b'$' => {
-            let len = read_decimal(buf, pos)?;
+            let len = read_decimal(buf, pos, ProtoFault::BulkLen)?;
             if len == -1 {
                 return Ok(());
             } // Null bulk string
             if len < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::BulkLen,
                     message: format!("invalid bulk string length: {}", len),
                     offset: *pos,
                 });
@@ -384,6 +416,7 @@ fn validate_frame(
             let len = len as usize;
             if len > config.max_bulk_string_size {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::BulkLen,
                     message: format!(
                         "bulk string size {} exceeds maximum {}",
                         len, config.max_bulk_string_size
@@ -400,9 +433,10 @@ fn validate_frame(
         }
         b'=' => {
             // VerbatimString: length-prefixed like bulk string
-            let len = read_decimal(buf, pos)?;
+            let len = read_decimal(buf, pos, ProtoFault::BulkLen)?;
             if len < 4 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!("verbatim string length {} too short", len),
                     offset: *pos,
                 });
@@ -410,6 +444,7 @@ fn validate_frame(
             let len = len as usize;
             if len > config.max_bulk_string_size {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!(
                         "verbatim string size {} exceeds maximum {}",
                         len, config.max_bulk_string_size
@@ -423,6 +458,7 @@ fn validate_frame(
             }
             if buf[*pos + 3] != b':' {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: "verbatim string missing ':' after 3-byte encoding".into(),
                     offset: *pos + 3,
                 });
@@ -432,12 +468,25 @@ fn validate_frame(
         }
         b'*' | b'~' | b'>' => {
             // Array, Set, Push: count + elements
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count == -1 {
                 return Ok(());
             } // Null array
             if count < 0 {
+                // Below -1 is lenient for `*` ONLY. Measured against
+                // redis-server 8.6.1: `*-9\r\n` is consumed silently and the
+                // connection keeps serving, where Moon used to kill it.
+                //
+                // Scoped to `*` deliberately: RESP3 Set (`~`) and Push (`>`)
+                // have no such Redis behaviour to match, and blanket leniency
+                // silently stopped rejecting `~-2` — caught by
+                // `test_resp3_negative_set_count`, which is why that test
+                // exists.
+                if type_byte == b'*' {
+                    return Ok(());
+                }
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid array/set/push length: {}", count),
                     offset: *pos,
                 });
@@ -445,6 +494,7 @@ fn validate_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -459,12 +509,13 @@ fn validate_frame(
         }
         b'%' => {
             // Map: count pairs
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count == -1 {
                 return Ok(()); // Null map
             }
             if count < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid map length: {}", count),
                     offset: *pos,
                 });
@@ -472,6 +523,7 @@ fn validate_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "map length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -486,6 +538,7 @@ fn validate_frame(
             Ok(())
         }
         byte => Err(ParseError::Invalid {
+            kind: ProtoFault::UnknownType(byte),
             message: format!("unknown type byte: 0x{:02x}", byte),
             offset: *pos - 1,
         }),
@@ -504,6 +557,7 @@ fn parse_single_frame(
 ) -> Result<Frame, ParseError> {
     if depth > config.max_array_depth {
         return Err(ParseError::Invalid {
+            kind: ProtoFault::MultibulkLen,
             message: format!(
                 "array nesting depth {} exceeds maximum {}",
                 depth, config.max_array_depth
@@ -534,6 +588,7 @@ fn parse_single_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             let line = &buf[*pos..crlf];
             let n = strict_atoi(line).ok_or_else(|| ParseError::Invalid {
+                kind: ProtoFault::ExpectedDollar(type_byte),
                 message: format!("invalid integer: {:?}", String::from_utf8_lossy(line)),
                 offset: *pos,
             })?;
@@ -541,12 +596,13 @@ fn parse_single_frame(
             Ok(Frame::Integer(n))
         }
         b'$' => {
-            let len = read_decimal(buf, pos)?;
+            let len = read_decimal(buf, pos, ProtoFault::BulkLen)?;
             if len == -1 {
                 return Ok(Frame::Null);
             }
             if len < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::BulkLen,
                     message: format!("invalid bulk string length: {}", len),
                     offset: *pos,
                 });
@@ -554,6 +610,7 @@ fn parse_single_frame(
             let len = len as usize;
             if len > config.max_bulk_string_size {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::BulkLen,
                     message: format!(
                         "bulk string size {} exceeds maximum {}",
                         len, config.max_bulk_string_size
@@ -573,12 +630,13 @@ fn parse_single_frame(
             Ok(frame)
         }
         b'*' => {
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count == -1 {
                 return Ok(Frame::Null);
             }
             if count < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid array length: {}", count),
                     offset: *pos,
                 });
@@ -586,6 +644,7 @@ fn parse_single_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "array length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -605,6 +664,7 @@ fn parse_single_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             if crlf != *pos {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!(
                         "RESP3 null has trailing data before CRLF at offset {}",
                         *pos
@@ -624,6 +684,7 @@ fn parse_single_frame(
                 b"f" => false,
                 _ => {
                     return Err(ParseError::Invalid {
+                        kind: ProtoFault::ExpectedDollar(type_byte),
                         message: format!(
                             "invalid boolean value: {:?}",
                             String::from_utf8_lossy(line)
@@ -640,6 +701,7 @@ fn parse_single_frame(
             let crlf = find_crlf(buf, *pos).ok_or(ParseError::Incomplete)?;
             let line = &buf[*pos..crlf];
             let s = std::str::from_utf8(line).map_err(|_| ParseError::Invalid {
+                kind: ProtoFault::ExpectedDollar(type_byte),
                 message: "invalid UTF-8 in double".into(),
                 offset: *pos,
             })?;
@@ -651,6 +713,7 @@ fn parse_single_frame(
                 f64::NAN
             } else {
                 s.parse::<f64>().map_err(|_| ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!("invalid double: {:?}", s),
                     offset: *pos,
                 })?
@@ -667,9 +730,10 @@ fn parse_single_frame(
         }
         b'=' => {
             // RESP3 VerbatimString: `=<len>\r\n<enc>:<data>\r\n`
-            let len = read_decimal(buf, pos)?;
+            let len = read_decimal(buf, pos, ProtoFault::BulkLen)?;
             if len < 4 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!(
                         "verbatim string length {} too short (minimum 4 for encoding + colon)",
                         len
@@ -680,6 +744,7 @@ fn parse_single_frame(
             let len = len as usize;
             if len > config.max_bulk_string_size {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: format!(
                         "verbatim string size {} exceeds maximum {}",
                         len, config.max_bulk_string_size
@@ -694,6 +759,7 @@ fn parse_single_frame(
             let payload = &buf[*pos..*pos + len];
             if payload[3] != b':' {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::ExpectedDollar(type_byte),
                     message: "verbatim string missing ':' after 3-byte encoding".into(),
                     offset: *pos + 3,
                 });
@@ -705,9 +771,10 @@ fn parse_single_frame(
         }
         b'%' => {
             // RESP3 Map: `%<count>\r\n<key><value>...`
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid map length: {}", count),
                     offset: *pos,
                 });
@@ -715,6 +782,7 @@ fn parse_single_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "map length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -732,9 +800,10 @@ fn parse_single_frame(
         }
         b'~' => {
             // RESP3 Set: `~<count>\r\n<elements...>`
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid set length: {}", count),
                     offset: *pos,
                 });
@@ -742,6 +811,7 @@ fn parse_single_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "set length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -757,9 +827,10 @@ fn parse_single_frame(
         }
         b'>' => {
             // RESP3 Push: `><count>\r\n<elements...>`
-            let count = read_decimal(buf, pos)?;
+            let count = read_decimal(buf, pos, ProtoFault::MultibulkLen)?;
             if count < 0 {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!("invalid push length: {}", count),
                     offset: *pos,
                 });
@@ -767,6 +838,7 @@ fn parse_single_frame(
             let count = count as usize;
             if count > config.max_array_length {
                 return Err(ParseError::Invalid {
+                    kind: ProtoFault::MultibulkLen,
                     message: format!(
                         "push length {} exceeds maximum {}",
                         count, config.max_array_length
@@ -781,6 +853,7 @@ fn parse_single_frame(
             Ok(Frame::Push(items))
         }
         byte => Err(ParseError::Invalid {
+            kind: ProtoFault::UnknownType(byte),
             message: format!("unknown type byte: 0x{:02x}", byte),
             offset: *pos - 1,
         }),
diff --git a/src/server/codec.rs b/src/server/codec.rs
index ca1fd8e8..726a3dc4 100644
--- a/src/server/codec.rs
+++ b/src/server/codec.rs
@@ -29,6 +29,10 @@ pub struct RespCodec {
     /// run during the growth that matters. `decode` is the one place that
     /// sees the buffer on every pass.
     max_query_buf: usize,
+    /// Fault from the last failed decode — see `take_last_fault`. The
+    /// `Decoder` trait's `io::Error` cannot carry it, and a bare close leaves
+    /// the client unable to tell a bad encoder from a dropped network.
+    last_fault: Option<crate::protocol::ProtoFault>,
 }
 
 impl RespCodec {
@@ -37,6 +41,7 @@ impl RespCodec {
             config,
             protocol_version: 2,
             max_query_buf: 0,
+            last_fault: None,
         }
     }
 
@@ -68,10 +73,29 @@ impl RespCodec {
         match protocol::parse(src, &self.config) {
             Ok(frame) => Ok(frame),
             Err(ParseError::Incomplete) => Ok(None),
-            Err(e) => Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
+            Err(e) => {
+                // `Decoder` pins the error type to `io::Error`, which cannot
+                // carry the fault kind — and the kind is exactly what the
+                // client needs to be told. Park it here for the handler to
+                // pick up; it is read immediately after this returns Err, so
+                // there is no window in which it can go stale.
+                if let ParseError::Invalid { kind, .. } = &e {
+                    self.last_fault = Some(*kind);
+                }
+                Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e))
+            }
         }
     }
 
+    /// The fault from the most recent failed [`Self::decode_frame`], taken.
+    ///
+    /// `None` means the read failed for a non-protocol reason (a dead socket,
+    /// the query-buffer ceiling) — in which case there is nothing useful to
+    /// tell the client and the connection just closes.
+    pub fn take_last_fault(&mut self) -> Option<crate::protocol::ProtoFault> {
+        self.last_fault.take()
+    }
+
     /// Encode a frame into the buffer (runtime-agnostic).
     ///
     /// Serializes using RESP2 or RESP3 format based on `protocol_version`.
diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs
index 18402a38..f8835720 100644
--- a/src/server/conn/core.rs
+++ b/src/server/conn/core.rs
@@ -263,6 +263,18 @@ pub(crate) struct ConnectionState {
     /// Set by WS.AUTH, cleared on connection drop.
     pub workspace_id: Option<WorkspaceId>,
     pub command_queue: Vec<Frame>,
+    /// The open transaction hit a QUEUE-TIME fault (unknown command, wrong
+    /// arity, an unqueueable verb) and must be refused wholesale at EXEC.
+    ///
+    /// Redis's `CLIENT_DIRTY_EXEC`. Without it Moon ran the half of a
+    /// transaction that happened to parse: `MULTI / NOSUCHCMD / SET k v /
+    /// EXEC` left `k` set, where Redis discards everything.
+    ///
+    /// Connection-local by construction, so shard count cannot affect it.
+    /// MUST be cleared on every exit from MULTI state — EXEC, DISCARD, RESET.
+    /// A leaked flag silently aborts an innocent later transaction, which is
+    /// a worse bug than the one it fixes.
+    pub multi_dirty: bool,
 
     /// c1M P1: this connection has issued REPLCONF — it is (almost
     /// certainly) a replica mid-handshake that will send PSYNC next, and
@@ -354,6 +366,7 @@ impl ConnectionState {
             active_cross_txn: None,
             workspace_id: migrated.and_then(|s| s.workspace_id),
             command_queue: Vec::new(),
+            multi_dirty: false,
             // Not carried through MigratedConnectionState: a conn that sent
             // REPLCONF never parks (so never rehydrates through this path),
             // and migration of a mid-handshake replica is not a supported
diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs
index 83ac7b8c..472b0a9c 100644
--- a/src/server/conn/handler_monoio/mod.rs
+++ b/src/server/conn/handler_monoio/mod.rs
@@ -98,6 +98,7 @@ use crate::command::metadata;
 use crate::command::{DispatchResult, dispatch, dispatch_read};
 use crate::persistence::aof;
 use crate::protocol::Frame;
+use crate::server::conn::shared::is_transaction_control;
 use crate::shard::dispatch::key_to_shard;
 use crate::shard::mesh::ChannelMesh;
 use crate::storage::eviction::{EvictionRun, evict_to_budget};
@@ -508,6 +509,10 @@ pub(crate) async fn handle_connection_sharded_monoio<
 
     // Pre-allocate frames Vec outside the loop; reused via .clear() each iteration.
     let mut frames: Vec<Frame> = Vec::with_capacity(64);
+    // Set when the parser rejects a frame. The connection still dies, but not
+    // silently and not before the valid frames that preceded the fault in the
+    // same read have been executed and answered.
+    let mut proto_fault: Option<crate::protocol::ProtoFault> = None;
 
     loop {
         // Check if CLIENT KILL targeted this connection (lock-free, QW8)
@@ -1222,11 +1227,26 @@ pub(crate) async fn handle_connection_sharded_monoio<
                     }
                 }
                 Ok(None) => break,
-                Err(_) => return (MonoioHandlerResult::Done, None), // parse error, close connection
+                Err(_) => {
+                    // A protocol fault kills the connection, but not before
+                    // the frames already parsed from this same read are run
+                    // and answered, and not before the client is told WHY.
+                    // The old `return Done` here dropped both — so
+                    // `PING\r\n<bad frame>` in one write answered nothing.
+                    proto_fault = codec.take_last_fault();
+                    break;
+                }
             }
         }
 
         if frames.is_empty() {
+            // Nothing valid preceded the fault: report it and close.
+            if let Some(kind) = proto_fault.take() {
+                let data = bytes::Bytes::from(super::util::proto_error_frame(kind));
+                let (_wr, _b): (std::io::Result<usize>, bytes::Bytes) =
+                    stream.write_all(data).await;
+                return (MonoioHandlerResult::Done, None);
+            }
             continue;
         }
 
@@ -1448,6 +1468,59 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 continue;
             }
 
+            // === MULTI QUEUE GATE — every non-transactional intercept MUST
+            // sit below this ===
+            //
+            // Redis executes exactly six commands while a transaction is open
+            // (MULTI, EXEC, DISCARD, WATCH, RESET, QUIT); EVERYTHING else
+            // queues. Moon decided this hundreds of lines further down, below
+            // the INFO / CLIENT / WS / MQ / PUBLISH / SUBSCRIBE intercepts, so
+            // each of those executed for real inside a transaction. Measured
+            // against redis-server 8.6.1: `SUBSCRIBE ch` put the connection
+            // into subscriber mode mid-MULTI, and `INFO server` returned a
+            // 3 KB dump where Redis returns `+QUEUED`.
+            //
+            // Deliberately BELOW the ACL gate: Redis refuses a forbidden
+            // command at queue time, and this repo has already shipped one ACL
+            // bypass that came from an intercept sitting above its check.
+            //
+            // This handler is the DEFAULT runtime (`runtime-monoio`). The
+            // sharded/tokio handlers carry the identical gate — a fix in only
+            // one of them is invisible to whichever job builds the other.
+            if conn.in_multi
+                && !is_transaction_control(cmd)
+                && !crate::server::conn::shared::is_intercept_only(cmd)
+            {
+                // FT.* isn't wired through the txn execution path; reject it
+                // explicitly rather than failing incidentally later.
+                if cmd.len() > 3 && cmd[..3].eq_ignore_ascii_case(b"FT.") {
+                    responses.push(Frame::Error(Bytes::from_static(
+                        b"ERR FT.* commands are not supported inside MULTI/EXEC",
+                    )));
+                    continue;
+                }
+                // Queue-time validation (Redis CLIENT_DIRTY_EXEC): a command
+                // that could never run poisons the whole transaction HERE, so
+                // EXEC refuses everything rather than applying the valid half.
+                if let Some(err) = crate::server::conn::shared::queue_time_rejection(cmd, cmd_args)
+                {
+                    conn.multi_dirty = true;
+                    responses.push(err);
+                    continue;
+                }
+                // Blocking commands queue as their non-blocking twin: a raw
+                // BLPOP reaching EXEC would BLOCK inside the transaction — a
+                // hang, not a divergence.
+                if crate::server::conn::blocking::is_blocking_command(cmd) {
+                    conn.command_queue
+                        .push(convert_blocking_to_nonblocking(cmd, cmd_args));
+                } else {
+                    conn.command_queue.push(frame);
+                }
+                responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED")));
+                continue;
+            }
+
             if cmd_len == 7 && dispatch::try_handle_cluster(cmd, cmd_args, ctx, &mut responses) {
                 continue;
             }
@@ -3185,6 +3258,14 @@ pub(crate) async fn handle_connection_sharded_monoio<
             break;
         }
 
+        // The valid prefix has been executed and flushed. Name the fault,
+        // then close — in that order.
+        if let Some(kind) = proto_fault.take() {
+            let data = bytes::Bytes::from(super::util::proto_error_frame(kind));
+            let (_wr, _b): (std::io::Result<usize>, bytes::Bytes) = stream.write_all(data).await;
+            return (MonoioHandlerResult::Done, None);
+        }
+
         // Check shutdown (polled after each batch -- acceptable for MVP)
         if shutdown.is_cancelled() {
             break;
diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs
index c488a226..7034020e 100644
--- a/src/server/conn/handler_monoio/write.rs
+++ b/src/server/conn/handler_monoio/write.rs
@@ -742,6 +742,15 @@ pub(super) async fn try_handle_multi_exec(
             responses.push(Frame::Error(Bytes::from_static(b"ERR EXEC without MULTI")));
         } else {
             conn.in_multi = false;
+            // A transaction poisoned at queue time executes NOTHING. Redis's
+            // CLIENT_DIRTY_EXEC: the client was already told which command was
+            // bad; EXEC's job is to make sure none of the others ran.
+            if std::mem::take(&mut conn.multi_dirty) {
+                conn.command_queue.clear();
+                conn.watched_keys.clear();
+                responses.push(crate::server::conn::shared::execabort_frame());
+                return true;
+            }
             // Taken, not borrowed: EXEC must clear its watches on BOTH the
             // committed and the aborted outcome, and a stale watch surviving
             // an abort is how a CAS retry loop livelocks.
@@ -1021,6 +1030,9 @@ pub(super) async fn try_handle_multi_exec(
         } else {
             conn.in_multi = false;
             conn.command_queue.clear();
+            // DISCARD must clear the poison too, or the NEXT transaction on
+            // this connection aborts for a fault that was never its own.
+            conn.multi_dirty = false;
             conn.watched_keys.clear();
             responses.push(Frame::SimpleString(Bytes::from_static(b"OK")));
         }
diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs
index 84e41d70..1b6e60dd 100644
--- a/src/server/conn/handler_sharded/mod.rs
+++ b/src/server/conn/handler_sharded/mod.rs
@@ -72,6 +72,7 @@ use crate::command::metadata;
 use crate::command::{DispatchResult, dispatch, dispatch_read};
 use crate::persistence::aof;
 use crate::protocol::Frame;
+use crate::server::conn::shared::is_transaction_control;
 use crate::shard::dispatch::{ShardMessage, key_to_shard};
 use crate::shard::mesh::ChannelMesh;
 use crate::storage::eviction::{EvictionRun, evict_to_budget};
@@ -442,6 +443,10 @@ pub(crate) async fn handle_connection_sharded_inner<
     };
 
     let mut break_outer = false;
+    // Set when the parser rejects a frame. The connection still dies, but not
+    // silently and not before the valid frames that preceded the fault in the
+    // same read have been executed and answered — see the parse loop below.
+    let mut proto_fault: Option<crate::protocol::ProtoFault> = None;
     loop {
         // Check if CLIENT KILL targeted this connection (lock-free, QW8)
         if client_live.is_killed() {
@@ -539,11 +544,34 @@ pub(crate) async fn handle_connection_sharded_inner<
                         }
                         Ok(None) => break,
                         Err(crate::protocol::ParseError::Incomplete) => break,
-                        Err(_) => { break_outer = true; break; }
+                        // A protocol fault is fatal to the connection, but it
+                        // is not a licence to drop what the client already
+                        // sent. `batch` holds every VALID frame parsed from
+                        // this same read; the old `Err(_) => break` discarded
+                        // it along with the reason, so `PING\r\n<bad frame>`
+                        // in one write answered nothing at all and closed mute.
+                        // Record the fault, stop parsing, and let the normal
+                        // batch machinery below execute and flush the prefix.
+                        // The error frame and the close happen after that.
+                        Err(crate::protocol::ParseError::Invalid { kind, .. }) => {
+                            proto_fault = Some(kind);
+                            break;
+                        }
+                        // The socket is already gone; there is nobody to tell.
+                        Err(crate::protocol::ParseError::Io(_)) => { break_outer = true; break; }
                     }
                 }
                 if break_outer { break; }
-                if batch.is_empty() { continue; }
+                if batch.is_empty() {
+                    // Nothing valid preceded the fault — report it directly.
+                    if let Some(kind) = proto_fault.take() {
+                        let _ = stream
+                            .write_all(super::util::proto_error_frame(kind).as_bytes())
+                            .await;
+                        break;
+                    }
+                    continue;
+                }
 
                 // CLIENT PAUSE: delay processing if server is paused
                 // Check with is_write=true (conservative — pauses all batches in ALL mode)
@@ -822,6 +850,62 @@ pub(crate) async fn handle_connection_sharded_inner<
                         }
                     }
 
+                    // === MULTI QUEUE GATE — every non-transactional intercept
+                    // MUST sit below this ===
+                    //
+                    // Redis executes exactly six commands while a transaction
+                    // is open (MULTI, EXEC, DISCARD, WATCH, RESET, QUIT);
+                    // EVERYTHING else queues. Moon decided this ~500 lines
+                    // further down, below the INFO / CLIENT / WS / MQ /
+                    // PUBLISH / SUBSCRIBE intercepts, so each of those ran for
+                    // real inside a transaction. Measured against
+                    // redis-server 8.6.1: `SUBSCRIBE ch` put the connection
+                    // into subscriber mode mid-MULTI, and `INFO server`
+                    // returned a 3 KB dump where Redis returns `+QUEUED`.
+                    //
+                    // Deliberately BELOW the ACL gate: Redis refuses a
+                    // forbidden command at queue time, and this repo has
+                    // already shipped one ACL bypass that came from an
+                    // intercept sitting above its permission check.
+                    //
+                    // The replication handshake verbs are exempt. A replica
+                    // never opens a transaction, so queueing them could only
+                    // ever break a handshake — and no test covers that.
+                    if conn.in_multi
+                        && !is_transaction_control(cmd)
+                        && !crate::server::conn::shared::is_intercept_only(cmd)
+                    {
+                        // FT.* vector commands aren't wired through the txn
+                        // execution path; reject them explicitly inside MULTI
+                        // (matches handler_single) rather than failing later.
+                        if cmd.len() > 3 && cmd[..3].eq_ignore_ascii_case(b"FT.") {
+                            responses.push(Frame::Error(Bytes::from_static(
+                                b"ERR FT.* commands are not supported inside MULTI/EXEC",
+                            )));
+                            continue;
+                        }
+                        // Queue-time validation (Redis CLIENT_DIRTY_EXEC): a
+                        // command that could never run poisons the whole
+                        // transaction HERE, so EXEC refuses everything rather
+                        // than applying the half that happened to be valid.
+                        if let Some(err) = crate::server::conn::shared::queue_time_rejection(cmd, cmd_args) {
+                            conn.multi_dirty = true;
+                            responses.push(err);
+                            continue;
+                        }
+                        // Blocking commands queue as their non-blocking twin:
+                        // a raw BLPOP reaching EXEC would BLOCK inside the
+                        // transaction — a hang, not a divergence.
+                        if crate::server::conn::blocking::is_blocking_command(cmd) {
+                            conn.command_queue
+                                .push(convert_blocking_to_nonblocking(cmd, cmd_args));
+                        } else {
+                            conn.command_queue.push(frame);
+                        }
+                        responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED")));
+                        continue;
+                    }
+
                     // --- CLUSTER subcommands ---
                     if cmd.eq_ignore_ascii_case(b"CLUSTER") {
                         if let Some(ref cs) = ctx.cluster_state {
@@ -1176,12 +1260,6 @@ pub(crate) async fn handle_connection_sharded_inner<
 
                     // --- BLOCKING COMMANDS ---
                     if crate::server::conn::blocking::is_blocking_command(cmd) {
-                        if conn.in_multi {
-                            let nb_frame = convert_blocking_to_nonblocking(cmd, cmd_args);
-                            conn.command_queue.push(nb_frame);
-                            responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED")));
-                            continue;
-                        }
                         // Earlier frames in this batch may hold barrier-pending
                         // local-leg writes — confirm (or fail-loud) them before
                         // this early flush; the replacement of `responses` below
@@ -1287,22 +1365,6 @@ pub(crate) async fn handle_connection_sharded_inner<
                         continue;
                     }
 
-                    // --- MULTI queue mode ---
-                    if conn.in_multi {
-                        // FT.* vector commands aren't wired through the txn
-                        // execution path; reject them explicitly inside MULTI
-                        // (matches handler_single) instead of an incidental
-                        // later error.
-                        if cmd.len() > 3 && cmd[..3].eq_ignore_ascii_case(b"FT.") {
-                            responses.push(Frame::Error(Bytes::from_static(
-                                b"ERR FT.* commands are not supported inside MULTI/EXEC",
-                            )));
-                            continue;
-                        }
-                        conn.command_queue.push(frame);
-                        responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED")));
-                        continue;
-                    }
 
                     // --- BGSAVE / SAVE / LASTSAVE / BGREWRITEAOF ---
                     if dispatch::try_handle_persistence(cmd, ctx, &mut responses) {
@@ -2571,6 +2633,16 @@ pub(crate) async fn handle_connection_sharded_inner<
                 }
 
                 if should_quit { break; }
+
+                // The valid prefix has now been executed and flushed. Name the
+                // fault, then close — in that order, so the client can tell a
+                // bad encoder from a dropped network.
+                if let Some(kind) = proto_fault.take() {
+                    let _ = stream
+                        .write_all(super::util::proto_error_frame(kind).as_bytes())
+                        .await;
+                    break;
+                }
             }
             // CLIENT TRACKING: deliver invalidation Push frames while the
             // connection is idle in read(). Pending-branch when tracking is
diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs
index 9553949b..1d78cf4f 100644
--- a/src/server/conn/handler_sharded/write.rs
+++ b/src/server/conn/handler_sharded/write.rs
@@ -567,6 +567,15 @@ pub(super) async fn try_handle_multi_exec(
             responses.push(Frame::Error(Bytes::from_static(b"ERR EXEC without MULTI")));
         } else {
             conn.in_multi = false;
+            // A transaction poisoned at queue time executes NOTHING. Redis's
+            // CLIENT_DIRTY_EXEC: the client was already told which command was
+            // bad; EXEC's job is to make sure none of the others ran.
+            if std::mem::take(&mut conn.multi_dirty) {
+                conn.command_queue.clear();
+                conn.watched_keys.clear();
+                responses.push(crate::server::conn::shared::execabort_frame());
+                return true;
+            }
             // Taken, not borrowed: EXEC must clear its watches on BOTH the
             // committed and the aborted outcome, and a stale watch surviving
             // an abort is how a CAS retry loop livelocks.
@@ -799,6 +808,9 @@ pub(super) async fn try_handle_multi_exec(
         } else {
             conn.in_multi = false;
             conn.command_queue.clear();
+            // DISCARD must clear the poison too, or the NEXT transaction on
+            // this connection aborts for a fault that was never its own.
+            conn.multi_dirty = false;
             conn.watched_keys.clear();
             responses.push(Frame::SimpleString(Bytes::from_static(b"OK")));
         }
diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs
index 046745a8..bb849a7d 100644
--- a/src/server/conn/handler_single.rs
+++ b/src/server/conn/handler_single.rs
@@ -495,16 +495,33 @@ pub async fn handle_connection(
             first_result = framed.next() => {
                 let first_frame = match first_result {
                     Some(Ok(frame)) => frame,
-                    Some(Err(_)) => break,
+                    Some(Err(_)) => {
+                        // Name the fault before closing. Nothing valid
+                        // preceded it — this IS the first frame of the read.
+                        if let Some(kind) = framed.codec_mut().take_last_fault() {
+                            use tokio::io::AsyncWriteExt;
+                            let msg = crate::server::conn::util::proto_error_frame(kind);
+                            let _ = framed.get_mut().write_all(msg.as_bytes()).await;
+                        }
+                        break;
+                    }
                     None => break,
                 };
 
                 // Collect batch: first frame + all immediately available frames
                 let mut batch = vec![first_frame];
                 const MAX_BATCH: usize = 1024;
+                // Set when a later frame in this same read is malformed: the
+                // batch collected so far still runs and is answered, and the
+                // fault is reported after it — never instead of it.
+                let mut proto_fault: Option<crate::protocol::ProtoFault> = None;
                 while batch.len() < MAX_BATCH {
                     match framed.next().now_or_never() {
                         Some(Some(Ok(frame))) => batch.push(frame),
+                        Some(Some(Err(_))) => {
+                            proto_fault = framed.codec_mut().take_last_fault();
+                            break;
+                        }
                         _ => break,
                     }
                 }
@@ -1411,6 +1428,16 @@ pub async fn handle_connection(
                                 ));
                             } else {
                                 conn.in_multi = false;
+                                // A transaction poisoned at queue time
+                                // executes NOTHING (Redis CLIENT_DIRTY_EXEC).
+                                if std::mem::take(&mut conn.multi_dirty) {
+                                    conn.command_queue.clear();
+                                    conn.watched_keys.clear();
+                                    responses.push(
+                                        crate::server::conn::shared::execabort_frame(),
+                                    );
+                                    continue;
+                                }
                                 let mut exec_publishes: Vec<(usize, Bytes, Bytes)> = Vec::new();
                                 // PR #282 review: `execute_transaction` holds
                                 // ONE guard on the db selected at EXEC time —
@@ -1580,6 +1607,9 @@ pub async fn handle_connection(
                             } else {
                                 conn.in_multi = false;
                                 conn.command_queue.clear();
+                                // DISCARD clears the poison too, or the NEXT
+                                // transaction aborts for a fault not its own.
+                                conn.multi_dirty = false;
                                 conn.watched_keys.clear();
                                 responses.push(Frame::SimpleString(Bytes::from_static(b"OK")));
                             }
@@ -1733,6 +1763,18 @@ pub async fn handle_connection(
                                 continue;
                             }
                         }
+                        // Queue-time validation (Redis CLIENT_DIRTY_EXEC): a
+                        // command that could never run poisons the whole
+                        // transaction HERE, so EXEC refuses everything rather
+                        // than applying the half that happened to be valid.
+                        if let Some((cmd, cmd_args)) = extract_command(&frame)
+                            && let Some(err) =
+                                crate::server::conn::shared::queue_time_rejection(cmd, cmd_args)
+                        {
+                            conn.multi_dirty = true;
+                            responses.push(err);
+                            continue;
+                        }
                         conn.command_queue.push(frame);
                         responses.push(Frame::SimpleString(Bytes::from_static(b"QUEUED")));
                         continue;
@@ -2869,6 +2911,15 @@ pub async fn handle_connection(
                 if break_outer || should_quit {
                     break;
                 }
+
+                // The valid prefix has been executed and flushed. Name the
+                // fault, then close — in that order.
+                if let Some(kind) = proto_fault.take() {
+                    use tokio::io::AsyncWriteExt;
+                    let msg = crate::server::conn::util::proto_error_frame(kind);
+                    let _ = framed.get_mut().write_all(msg.as_bytes()).await;
+                    break;
+                }
             }
             // Deliver tracking invalidation Push frames to client
             msg = async {
diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs
index ad6e2289..0c0e75bc 100644
--- a/src/server/conn/shared.rs
+++ b/src/server/conn/shared.rs
@@ -991,6 +991,181 @@ impl PubSubTeardown for parking_lot::Mutex<crate::pubsub::PubSubRegistry> {
     }
 }
 
+/// Commands that EXECUTE while a transaction is open instead of queueing.
+///
+/// Redis's list is exactly six: MULTI, EXEC, DISCARD, WATCH, RESET, QUIT.
+/// (WATCH executes only to be refused — `queue_time_rejection` returns its
+/// error — but it must reach that intercept rather than land in the queue.)
+///
+/// The replication handshake verbs are added on top, and only those. A replica
+/// never opens a transaction, so queueing REPLCONF or PSYNC could not fix any
+/// real client and could break a handshake no test covers — the asymmetry is
+/// the whole argument for the exemption.
+pub(crate) fn is_transaction_control(cmd: &[u8]) -> bool {
+    const CONTROL: [&[u8]; 12] = [
+        b"MULTI",
+        b"EXEC",
+        b"DISCARD",
+        b"WATCH",
+        b"UNWATCH",
+        b"RESET",
+        b"QUIT",
+        b"REPLCONF",
+        b"PSYNC",
+        b"SYNC",
+        b"REPLICAOF",
+        b"SLAVEOF",
+    ];
+    CONTROL.iter().any(|c| cmd.eq_ignore_ascii_case(c))
+}
+
+/// Commands the transaction executor CANNOT run, so queueing them would turn a
+/// working command into a guaranteed error inside `EXEC`.
+///
+/// `execute_transaction` replays the queue through `dispatch()`. These commands
+/// are connection-level *intercepts* — they never reach `dispatch()` at all, so
+/// a queued one comes back as `-ERR unknown command`. Measured 2026-08-12 by
+/// queueing each inside a MULTI and reading the EXEC array:
+///
+/// ```text
+/// CONFIG · CLIENT · ACL · CLUSTER · SCRIPT · WAIT          -> unknown command
+/// INFO · SLOWLOG · PUBLISH · MEMORY · DEBUG · COMMAND · OBJECT -> execute fine
+/// ```
+///
+/// Redis queues all of them and runs them properly. Moon cannot yet, so these
+/// keep EXECUTING immediately — the pre-existing divergence, which the
+/// client-compat manifest already waives — rather than queueing into a hard
+/// error. That is strictly better than the alternative: before the queue gate a
+/// client got its data with the wrong reply shape; queueing it unconditionally
+/// gave them an error instead, which is a regression, not a narrowing.
+///
+/// `LATENCY` looked like a member and is NOT: it errors outside a transaction
+/// too, because Moon does not implement it at all. Exempting it would have
+/// papered over nothing and hidden a genuine unimplemented command.
+///
+/// `PUBSUB` is absent from `COMMAND_META`, so `queue_time_rejection` would call
+/// it an unknown command and poison the transaction — it has no dot, so the
+/// dotted carve-out misses it. That is exactly the regression class the §1 ⚠
+/// assumption named. It is handled by `queue_time_rejection` consulting this
+/// list, not by exempting it from queueing.
+///
+/// Removing an entry from this list requires teaching `execute_transaction` to
+/// run it. The test `me10` asserts every queued command's EXEC result is not an
+/// error, so a premature removal fails loudly.
+pub(crate) fn is_intercept_only(cmd: &[u8]) -> bool {
+    const INTERCEPT_ONLY: [&[u8]; 7] = [
+        b"CONFIG", b"CLIENT", b"ACL", b"CLUSTER", b"SCRIPT", b"WAIT", b"PUBSUB",
+    ];
+    INTERCEPT_ONLY.iter().any(|c| cmd.eq_ignore_ascii_case(c))
+}
+
+/// Validate a command at QUEUE time, the way Redis does before storing it.
+///
+/// Returns `Some(error)` when the command must not be queued. The caller
+/// replies that error INSTEAD of `+QUEUED`, does not push the command, and
+/// sets `conn.multi_dirty` so `EXEC` refuses the whole block.
+///
+/// This is the difference between a transaction that is atomic with respect to
+/// typos and one that is not. Measured against redis-server 8.6.1:
+///
+/// ```text
+/// MULTI / NOSUCHCMD / SET k v / EXEC
+///   redis -> k UNSET   (EXECABORT, nothing ran)
+///   moon  -> k SET     (the valid half ran)
+/// ```
+///
+/// Shared by all THREE handlers, for the same reason `RESET` and `WATCH` are:
+/// a queue-time check that lands on one runtime and not the others is not a
+/// fix, and no single-runtime CI job would notice.
+///
+/// The arity/existence check reads `COMMAND_META` — the SAME table dispatch
+/// consults — so a command cannot become queueable-but-undispatchable, or the
+/// reverse. That symmetry is the whole safety argument for this function:
+/// rejecting a command that DOES dispatch would be a regression strictly worse
+/// than the bug being fixed.
+pub(crate) fn queue_time_rejection(cmd: &[u8], args: &[Frame]) -> Option<Frame> {
+    // Verbs that are refused rather than queued. Redis rejects these because
+    // they change the connection's mode, which a queued command cannot
+    // meaningfully do. Moon used to EXECUTE `SUBSCRIBE` immediately, putting
+    // the connection into subscriber mode in the middle of a transaction.
+    const UNQUEUEABLE: [&[u8]; 4] = [b"SUBSCRIBE", b"UNSUBSCRIBE", b"PSUBSCRIBE", b"PUNSUBSCRIBE"];
+    for verb in UNQUEUEABLE {
+        if cmd.eq_ignore_ascii_case(verb) {
+            let name = String::from_utf8_lossy(verb).to_uppercase();
+            return Some(Frame::Error(Bytes::from(format!(
+                "ERR {name} is not allowed in transactions"
+            ))));
+        }
+    }
+    if cmd.eq_ignore_ascii_case(b"WATCH") {
+        return Some(Frame::Error(Bytes::from_static(
+            b"ERR WATCH inside MULTI is not allowed",
+        )));
+    }
+
+    let Some(meta) = crate::command::metadata::lookup(cmd) else {
+        // Only a NON-namespaced name can confidently be called unknown.
+        //
+        // `COMMAND_META` (263 entries) covers Redis's surface plus the Moon
+        // extensions that were registered — `GRAPH.QUERY` is in it — but NOT
+        // every dotted family the handlers intercept ahead of dispatch.
+        // Rejecting an unregistered dotted name here would break a command
+        // that works fine OUTSIDE a transaction: a regression strictly worse
+        // than the bug this function exists to fix.
+        //
+        // Audited 2026-08-12: of the dotted families absent from the table,
+        // `TXN` is intercepted before the queue block, `FT.*` is rejected
+        // above it, and `TS.*` / `JSON.*` do not exist in Moon at all. So
+        // nothing is broken today — but that is four separate accidents, not
+        // a safety argument, and the next dotted family added without a table
+        // entry would silently become unusable inside MULTI.
+        //
+        // The carve-out costs only the ability to catch a typo'd dotted name.
+        // Dotted names that ARE registered still get their arity checked.
+        if cmd.contains(&b'.') {
+            return None;
+        }
+        // Redis's format: the name quoted, then the args it did get. A driver
+        // author reading this sees their typo; a generic "unknown command"
+        // sends them looking at Moon.
+        let mut detail = String::new();
+        for a in args.iter().take(20) {
+            if let Frame::BulkString(b) | Frame::SimpleString(b) = a {
+                detail.push('\'');
+                detail.push_str(&String::from_utf8_lossy(b));
+                detail.push_str("', ");
+            }
+        }
+        return Some(Frame::Error(Bytes::from(format!(
+            "ERR unknown command '{}', with args beginning with: {detail}",
+            String::from_utf8_lossy(cmd)
+        ))));
+    };
+
+    // `arity` counts the command name itself, so compare against args + 1.
+    // Positive = exact; negative = minimum (variadic).
+    let given = args.len() as i16 + 1;
+    let bad_arity = if meta.arity >= 0 {
+        given != meta.arity
+    } else {
+        given < -meta.arity
+    };
+    if bad_arity {
+        return Some(Frame::Error(Bytes::from(format!(
+            "ERR wrong number of arguments for '{}' command",
+            meta.name.to_lowercase()
+        ))));
+    }
+    None
+}
+
+/// The reply `EXEC` owes a transaction poisoned at queue time.
+pub(crate) fn execabort_frame() -> Frame {
+    Frame::Error(Bytes::from_static(
+        b"EXECABORT Transaction discarded because of previous errors.",
+    ))
+}
+
 /// Handle `RESET`, returning `true` when the command was consumed.
 ///
 /// MUST be called BEFORE the MULTI queueing step. Measured against
@@ -1038,6 +1213,9 @@ pub(crate) fn try_handle_reset(
     // Transaction
     conn.in_multi = false;
     conn.command_queue.clear();
+    // The dirty flag must not survive the transaction it belongs to; a leak
+    // would abort the NEXT, innocent one.
+    conn.multi_dirty = false;
     conn.watched_keys.clear();
 
     // Client-side caching
diff --git a/src/server/conn/util.rs b/src/server/conn/util.rs
index b93e6f7e..7bd9a74e 100644
--- a/src/server/conn/util.rs
+++ b/src/server/conn/util.rs
@@ -277,6 +277,19 @@ pub(crate) fn query_buf_exceeded(
 pub(crate) const QUERY_BUF_LIMIT_ERROR: &[u8] =
     b"-ERR Protocol error: query buffer limit reached\r\n";
 
+/// Render a parse fault as the RESP error line Redis would send, CRLF and all.
+///
+/// Every handler funnels through this so the three of them cannot drift: a
+/// protocol fault that reads one way on monoio and another on tokio is a
+/// compatibility bug that no single-runtime CI job would ever catch.
+///
+/// Allocates. That is deliberate and safe: this is reached once per doomed
+/// connection, never on a serving path, so it is not a hot-path allocation in
+/// the sense the coding rules police.
+pub(crate) fn proto_error_frame(kind: crate::protocol::ProtoFault) -> String {
+    format!("-ERR {}\r\n", kind.wire_text_owned())
+}
+
 #[cfg(test)]
 mod query_buf_limit_tests {
     use super::*;
diff --git a/tests/multi_exec_queue_semantics.rs b/tests/multi_exec_queue_semantics.rs
new file mode 100644
index 00000000..a12dbd8a
--- /dev/null
+++ b/tests/multi_exec_queue_semantics.rs
@@ -0,0 +1,571 @@
+//! A transaction must be all-or-nothing with respect to QUEUE-TIME faults.
+//!
+//! Measured against redis-server 8.6.1. The headline is a data-correctness
+//! bug, not a compatibility nit:
+//!
+//! ```text
+//! MULTI / NOSUCHCMD / SET k v / EXEC
+//!   redis: -ERR unknown command … / +QUEUED / -EXECABORT      -> k is UNSET
+//!   moon : -ERR unknown command … / +QUEUED / *1 +OK          -> k is SET
+//! ```
+//!
+//! Moon has no queue-time validation, so nothing marks the transaction dirty
+//! and EXEC happily runs the half that parsed. Redis checks each command's
+//! existence and arity as it is QUEUED, latches a dirty flag on failure, and
+//! refuses the whole block at EXEC.
+//!
+//! Raw sockets throughout: redis-rs normalises away the very distinctions
+//! under test (a Null Array vs a Null Bulk, `+QUEUED` vs an error reply).
+
+mod common;
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+const EXECABORT: &str = "EXECABORT Transaction discarded because of previous errors.";
+
+struct Moon {
+    child: Child,
+    port: u16,
+    tmp_dir: std::path::PathBuf,
+}
+
+impl Drop for Moon {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        let _ = std::fs::remove_dir_all(&self.tmp_dir);
+    }
+}
+
+fn spawn_moon(shards: &str) -> Moon {
+    let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon"));
+    let (child, port) = common::spawn_listening(|port| {
+        let tmp_dir = std::env::temp_dir().join(format!("moon-multiexec-{port}"));
+        let _ = std::fs::create_dir_all(&tmp_dir);
+        Command::new(&bin)
+            .args([
+                "--port",
+                &port.to_string(),
+                "--shards",
+                shards,
+                "--admin-port",
+                "0",
+                "--appendonly",
+                "no",
+                "--disk-free-min-pct",
+                "0",
+                "--dir",
+                tmp_dir.to_str().unwrap(),
+            ])
+            .stdout(Stdio::null())
+            .stderr(
+                std::fs::File::create(tmp_dir.join("moon.stderr")).expect("create moon stderr log"),
+            )
+            .spawn()
+            .expect("spawn moon")
+    });
+    let tmp_dir = std::env::temp_dir().join(format!("moon-multiexec-{port}"));
+    let mut moon = Moon {
+        child,
+        port,
+        tmp_dir,
+    };
+    let deadline = Instant::now() + Duration::from_secs(30);
+    while Instant::now() < deadline {
+        if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) {
+            let _ = c.set_read_timeout(Some(Duration::from_millis(500)));
+            if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() {
+                let mut buf = [0u8; 64];
+                if let Ok(n) = c.read(&mut buf)
+                    && n > 0
+                    && buf.starts_with(b"+PONG")
+                {
+                    return moon;
+                }
+            }
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    let status = match moon.child.try_wait() {
+        Ok(Some(s)) => format!("exited with {s}"),
+        Ok(None) => "still running but never answered PING".to_string(),
+        Err(e) => format!("status unavailable: {e}"),
+    };
+    let log = std::fs::read_to_string(moon.tmp_dir.join("moon.stderr")).unwrap_or_default();
+    panic!("moon never became ready on port {port} ({status})\n--- stderr ---\n{log}");
+}
+
+/// A connection that sends one command at a time and returns each raw reply,
+/// so command N's answer is never conflated with command N+1's.
+struct Conn(TcpStream);
+
+impl Conn {
+    fn open(port: u16) -> Self {
+        let s = TcpStream::connect(("127.0.0.1", port)).expect("connect");
+        s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+        s.set_write_timeout(Some(Duration::from_secs(5))).unwrap();
+        Conn(s)
+    }
+
+    /// Encode as a RESP array. Inline would work for most of these, but the
+    /// inline path is a DIFFERENT dispatch path in Moon, and a MULTI fix that
+    /// only lands on one of them is not a fix.
+    fn send(&mut self, parts: &[&str]) -> String {
+        let mut out = format!("*{}\r\n", parts.len());
+        for p in parts {
+            out.push_str(&format!("${}\r\n{p}\r\n", p.len()));
+        }
+        self.0.write_all(out.as_bytes()).expect("write");
+        self.read_reply()
+    }
+
+    /// Send raw bytes with no encoding — for the malformed-frame case.
+    fn send_raw(&mut self, bytes: &[u8]) -> String {
+        let _ = self.0.write_all(bytes);
+        self.read_reply()
+    }
+
+    fn read_reply(&mut self) -> String {
+        let mut buf = [0u8; 8192];
+        let mut acc = Vec::new();
+        loop {
+            match self.0.read(&mut buf) {
+                Ok(0) => break,
+                Ok(n) => {
+                    acc.extend_from_slice(&buf[..n]);
+                    // One reply per send, so the first complete line (or the
+                    // whole array) is enough; a short second read drains any
+                    // trailing segment of a multi-line array.
+                    self.0
+                        .set_read_timeout(Some(Duration::from_millis(200)))
+                        .unwrap();
+                }
+                Err(_) => break,
+            }
+        }
+        self.0
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        String::from_utf8_lossy(&acc).into_owned()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// me1 — the data-loss case. This is why the task exists.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me1_unknown_command_aborts_and_applies_nothing() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    assert!(c.send(&["MULTI"]).starts_with("+OK"));
+
+    let bad = c.send(&["NOSUCHCMD", "a", "b"]);
+    assert!(
+        bad.contains("unknown command"),
+        "queueing an unknown command must reply the error AT QUEUE TIME; got {bad:?}"
+    );
+
+    let queued = c.send(&["SET", "me1key", "v"]);
+    assert!(
+        queued.starts_with("+QUEUED"),
+        "a VALID command after the bad one still queues normally; got {queued:?}"
+    );
+
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        exec.contains(EXECABORT),
+        "EXEC on a poisoned transaction must refuse the whole block; got {exec:?}"
+    );
+
+    // The assertion that makes this a correctness bug rather than a wording
+    // nit: on Moon today the SET actually lands.
+    let mut probe = Conn::open(m.port);
+    let got = probe.send(&["GET", "me1key"]);
+    assert!(
+        got.starts_with("$-1"),
+        "an aborted transaction must apply NOTHING. me1key should not exist; \
+         GET returned {got:?}"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me2 — arity is checked at queue time too.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me2_wrong_arity_aborts() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    let bad = c.send(&["GET"]);
+    assert!(
+        bad.contains("wrong number of arguments"),
+        "GET with no key cannot ever run; it must be refused at queue time. got {bad:?}"
+    );
+    let exec = c.send(&["EXEC"]);
+    assert!(exec.contains(EXECABORT), "got {exec:?}");
+}
+
+// ---------------------------------------------------------------------------
+// me3 — the dirty flag must not outlive its transaction.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me3_discard_clears_the_dirty_flag() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    c.send(&["NOSUCHCMD"]);
+    assert!(c.send(&["DISCARD"]).starts_with("+OK"));
+
+    // A leaked flag would silently abort this innocent transaction — a worse
+    // bug than the one being fixed, so it gets its own test.
+    c.send(&["MULTI"]);
+    assert!(c.send(&["SET", "me3key", "v"]).starts_with("+QUEUED"));
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        !exec.contains("EXECABORT") && exec.contains("+OK"),
+        "a fresh MULTI after a DISCARDed poisoned one must run normally; got {exec:?}"
+    );
+    assert!(c.send(&["GET", "me3key"]).contains('v'));
+}
+
+// ---------------------------------------------------------------------------
+// me4 — SUBSCRIBE must be refused, not EXECUTED. Moon subscribes for real.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me4_subscribe_is_refused_at_queue_time_not_executed() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    let r = c.send(&["SUBSCRIBE", "ch"]);
+    assert!(
+        !r.contains("subscribe") || r.contains("not allowed"),
+        "SUBSCRIBE inside MULTI must be REFUSED, not run. A reply containing a \
+         real subscribe confirmation means the connection entered subscriber \
+         mode inside a transaction. got {r:?}"
+    );
+    assert!(
+        r.contains("not allowed in transactions"),
+        "expected Redis's refusal text; got {r:?}"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me5 — WATCH has its own distinct refusal.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me5_watch_inside_multi_is_refused() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    let r = c.send(&["WATCH", "k"]);
+    assert!(r.contains("WATCH inside MULTI is not allowed"), "got {r:?}");
+}
+
+// ---------------------------------------------------------------------------
+// me6 — a malformed frame inside MULTI still names itself (protocol contract).
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me6_bad_frame_inside_multi_names_itself_before_closing() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    let r = c.send_raw(b"*2\r\n$3\r\nGET\r\n$abc\r\nk\r\n");
+    assert!(
+        r.contains("Protocol error: invalid bulk length"),
+        "a malformed frame inside MULTI follows the same protocol-error \
+         contract as anywhere else: name it, THEN close. Moon closes bare \
+         today. got {r:?}"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me7 — the null TYPE matters to statically-typed clients.
+// ---------------------------------------------------------------------------
+
+#[test]
+#[ignore = "needs Frame::NullArray — RESP2 `*-1` is inexpressible in Moon today; \
+            filed as its own task, see the comment below"]
+fn me7_blpop_in_multi_returns_null_array_not_null_bulk() {
+    // IGNORED, not deleted, and not weakened: the divergence is real and
+    // measured (Moon `*1\r\n$-1\r\n`, Redis `*1\r\n*-1\r\n`).
+    //
+    // It cannot be fixed here. `Frame` has no null-array variant at all —
+    // `Frame::Null` serialises to `$-1` in RESP2, full stop — so this is not a
+    // MULTI bug: a plain `BLPOP key 1` that times out OUTSIDE a transaction
+    // returns the wrong type too, as does every command whose null shape is an
+    // array. The fix is a `Frame::NullArray` variant threaded through the ~14
+    // `Frame::Null` arms in `serialize.rs` / `resp3.rs`, which touches every
+    // reply path and deserves its own contract and review.
+    //
+    // Un-ignore this test in that task; it is already the right assertion.
+
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    assert!(c.send(&["BLPOP", "me7missing", "0"]).starts_with("+QUEUED"));
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        exec.contains("*-1"),
+        "a blocking command that finds nothing inside MULTI takes its \
+         zero-timeout path and returns a Null ARRAY (`*-1`). Moon returns a \
+         Null BULK (`$-1`), which a statically-typed client decodes as the \
+         wrong type. got {exec:?}"
+    );
+    assert!(
+        !exec.contains("$-1"),
+        "the reply must not be a Null Bulk; got {exec:?}"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me8 — a RUNTIME error is data, not a queue-time fault. Must NOT abort.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me8_wrongtype_at_execution_does_not_abort_the_block() {
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["LPUSH", "me8list", "v"]);
+    c.send(&["MULTI"]);
+    c.send(&["GET", "me8list"]);
+    c.send(&["SET", "me8other", "v"]);
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        exec.contains("WRONGTYPE"),
+        "the type error is one ELEMENT of the EXEC array; got {exec:?}"
+    );
+    assert!(
+        !exec.contains("EXECABORT"),
+        "a runtime error must NOT abort the transaction — only queue-time \
+         faults do. got {exec:?}"
+    );
+    assert!(
+        c.send(&["GET", "me8other"]).contains('v'),
+        "the sibling command must still have applied"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me9 — the cases already matching Redis must stay matching.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me9_already_matching_behavior_is_unchanged() {
+    let m = spawn_moon("1");
+
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    assert!(
+        c.send(&["MULTI"]).contains("MULTI calls can not be nested"),
+        "nested MULTI"
+    );
+    c.send(&["DISCARD"]);
+
+    let mut c = Conn::open(m.port);
+    assert!(
+        c.send(&["EXEC"]).contains("EXEC without MULTI"),
+        "EXEC without MULTI"
+    );
+    assert!(
+        c.send(&["DISCARD"]).contains("DISCARD without MULTI"),
+        "DISCARD without MULTI"
+    );
+
+    // Empty MULTI/EXEC is an empty array, not an error.
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    assert!(c.send(&["EXEC"]).starts_with("*0"), "empty MULTI/EXEC");
+
+    // SELECT inside MULTI queues and applies.
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    assert!(c.send(&["SELECT", "3"]).starts_with("+QUEUED"));
+    assert!(c.send(&["EXEC"]).contains("+OK"), "SELECT inside MULTI");
+
+    // RESET inside MULTI abandons the transaction.
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    c.send(&["SET", "me9key", "v"]);
+    assert!(c.send(&["RESET"]).starts_with("+RESET"));
+    assert!(
+        c.send(&["EXEC"]).contains("EXEC without MULTI"),
+        "RESET must have left MULTI state"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me10 — the §1 ⚠ mitigation. A command that dispatches must also queue.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me10_every_dispatchable_command_is_queueable() {
+    // The risk this fix introduces: queue-time validation consults
+    // COMMAND_META, but Moon has THREE dispatch paths. A command reachable
+    // through dispatch but MISSING from the table would now be rejected
+    // inside MULTI while still working outside it — a regression strictly
+    // worse than the bug being fixed.
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+
+    // One representative command per dispatch path, each with valid arity.
+    let representatives: [&[&str]; 6] = [
+        &["SET", "me10k", "v"],       // write path: dispatch
+        &["GET", "me10k"],            // read path: dispatch_read + inline fast path
+        &["HSET", "me10h", "f", "v"], // write, non-inline-eligible
+        &["TTL", "me10k"],            // read, non-inline
+        &["PING"],                    // no-key command
+        &["INFO", "server"],          // admin-ish, optional arg
+    ];
+    for cmd in representatives {
+        c.send(&["MULTI"]);
+        let r = c.send(cmd);
+        assert!(
+            r.starts_with("+QUEUED"),
+            "{cmd:?} dispatches fine outside MULTI, so it MUST queue inside \
+             one. A non-QUEUED reply means COMMAND_META disagrees with \
+             dispatch. got {r:?}"
+        );
+        let exec = c.send(&["EXEC"]);
+        assert!(
+            !exec.contains("EXECABORT"),
+            "{cmd:?} queued but the transaction was poisoned anyway: {exec:?}"
+        );
+        // The assertion this test was MISSING. `+QUEUED` plus "no EXECABORT"
+        // says the transaction was not poisoned — it says nothing about
+        // whether the command actually RAN. It did not: `execute_transaction`
+        // replays the queue through `dispatch()`, and a connection-level
+        // intercept never reaches dispatch, so it came back as
+        // `-ERR unknown command` INSIDE the EXEC array while both assertions
+        // above stayed happy. That is how a regression shipped past a green
+        // test — the check verified the signal expected, not the behaviour
+        // claimed.
+        assert!(
+            !exec.contains("unknown command"),
+            "{cmd:?} queued but could not be EXECUTED — a queued command that \
+             cannot run is worse than one that never queued, because the client \
+             gets an error where it used to get data. got {exec:?}"
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// me10b — the regression class the ⚠ assumption named, pinned.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me10b_unregistered_dotted_commands_still_queue() {
+    // Queue-time validation reads COMMAND_META, which does NOT cover every
+    // dotted extension family Moon dispatches. A dotted name missing from the
+    // table must fall through to queueing, never be called "unknown" — that
+    // would break a command that works fine outside a transaction.
+    //
+    // GRAPH.QUERY is IN the table, so it proves registered dotted names are
+    // unaffected; the synthetic name proves unregistered ones are not
+    // rejected at queue time. Whether the synthetic one errors at EXEC is not
+    // this test's business — only that it does not poison the transaction.
+    let m = spawn_moon("1");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    let r = c.send(&["SOMEFUTURE.CMD", "arg"]);
+    assert!(
+        r.starts_with("+QUEUED"),
+        "an unregistered DOTTED command must queue, not be rejected as          unknown — otherwise adding a dotted family without a COMMAND_META          entry silently makes it unusable inside MULTI. got {r:?}"
+    );
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        !exec.contains("EXECABORT"),
+        "an unregistered dotted command must not poison the transaction; got {exec:?}"
+    );
+
+    // A NON-dotted unknown name is still caught — the carve-out must not have
+    // disabled the check that me1 depends on.
+    let mut c2 = Conn::open(m.port);
+    c2.send(&["MULTI"]);
+    assert!(
+        c2.send(&["NOSUCHCMD"]).contains("unknown command"),
+        "the dotted carve-out must not weaken the plain unknown-command check"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// me10c — intercept-only commands must not be queued into a guaranteed error.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me10c_intercept_only_commands_are_not_queued_into_an_error() {
+    // LATENCY is deliberately NOT in this list: it errors outside a
+    // transaction too (Moon does not implement it), so including it would have
+    // hidden an unimplemented command behind a transaction exemption.
+    //
+    // These reach a connection-level intercept and never `dispatch()`, so the
+    // transaction executor cannot replay them. Redis queues and runs them all;
+    // Moon cannot yet, so it keeps executing them immediately — the
+    // pre-existing divergence the compat manifest waives.
+    //
+    // What must NEVER happen is the middle state: queueing them into an
+    // `-ERR unknown command` inside EXEC. That turns data into an error, which
+    // is strictly worse than the divergence it replaced.
+    let m = spawn_moon("1");
+    for cmd in [
+        &["CONFIG", "GET", "maxmemory"][..],
+        &["CLIENT", "GETNAME"][..],
+        &["ACL", "WHOAMI"][..],
+        &["CLUSTER", "INFO"][..],
+        &["SCRIPT", "EXISTS", "deadbeef"][..],
+        &["WAIT", "0", "0"][..],
+        // PUBSUB is the one that matters most here: it works outside MULTI
+        // (`*0`) but is ABSENT from COMMAND_META, so queue-time validation
+        // would call it unknown and poison the transaction. It has no dot,
+        // so the dotted carve-out does not cover it.
+        &["PUBSUB", "CHANNELS"][..],
+    ] {
+        let mut c = Conn::open(m.port);
+        c.send(&["MULTI"]);
+        let queued = c.send(cmd);
+        assert!(
+            !queued.contains("unknown command"),
+            "{cmd:?} works outside MULTI, so it must not be rejected as unknown \
+             inside one. got {queued:?}"
+        );
+        let exec = c.send(&["EXEC"]);
+        assert!(
+            !exec.contains("unknown command") && !exec.contains("EXECABORT"),
+            "{cmd:?} must not end up as an error inside EXEC. got {exec:?}"
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// me11 — shard count must not change transaction semantics.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn me11_queue_semantics_hold_at_four_shards() {
+    let m = spawn_moon("4");
+    let mut c = Conn::open(m.port);
+    c.send(&["MULTI"]);
+    c.send(&["NOSUCHCMD"]);
+    // Deliberately a key that hashes elsewhere than the connection's shard.
+    c.send(&["SET", "me11:crossshard:key", "v"]);
+    let exec = c.send(&["EXEC"]);
+    assert!(
+        exec.contains(EXECABORT),
+        "the dirty flag is connection-local, so shard count cannot matter; got {exec:?}"
+    );
+    let mut probe = Conn::open(m.port);
+    assert!(
+        probe
+            .send(&["GET", "me11:crossshard:key"])
+            .starts_with("$-1"),
+        "a cross-shard write in an aborted transaction must not apply either"
+    );
+}
diff --git a/tests/protocol_error_lifetime.rs b/tests/protocol_error_lifetime.rs
new file mode 100644
index 00000000..2e1ffd2f
--- /dev/null
+++ b/tests/protocol_error_lifetime.rs
@@ -0,0 +1,433 @@
+//! A malformed frame must end the connection the way Redis ends it.
+//!
+//! Measured against redis-server 8.6.1 (raw sockets, identical bytes). Moon's
+//! divergences all trace to ONE line — `handler_sharded/mod.rs`:
+//!
+//! ```text
+//! Err(crate::protocol::ParseError::Incomplete) => break,
+//! Err(_) => { break_outer = true; break; }
+//! ```
+//!
+//! That arm throws away two things: the parse error's *reason* (so the client
+//! gets a bare FIN and cannot tell a bad encoder from a dropped network), and
+//! `batch` — which already holds every VALID frame parsed before the bad one,
+//! so `PING\r\n*-9\r\n` in one write answers nothing at all.
+//!
+//! These tests speak the raw socket rather than going through redis-rs,
+//! because the entire subject is bytes-and-close behavior that a client
+//! library normalises away. Each case asserts THREE things, since any one
+//! alone hides the interesting half:
+//!   * the reply bytes,
+//!   * whether the server hung up,
+//!   * whether a following command still gets served.
+
+mod common;
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+/// Redis 8.6.1's verbatim wire text. Byte-compared, not substring-matched
+/// where the whole string is known — a near-miss ("invalid bulk string
+/// length") is exactly the bug this task exists to fix.
+const ERR_BULK_LEN: &str = "ERR Protocol error: invalid bulk length";
+const ERR_INLINE_TOO_BIG: &str = "ERR Protocol error: too big inline request";
+const ERR_UNBALANCED: &str = "ERR Protocol error: unbalanced quotes in request";
+
+struct Moon {
+    child: Child,
+    port: u16,
+    tmp_dir: std::path::PathBuf,
+}
+
+impl Drop for Moon {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        let _ = std::fs::remove_dir_all(&self.tmp_dir);
+    }
+}
+
+fn spawn_moon(shards: &str) -> Moon {
+    // CARGO_BIN_EXE_moon is the binary cargo built for THIS invocation —
+    // fresh and feature-matched. Never probe target/release directly: that
+    // path's provenance is unknown and has produced false PASSes before.
+    let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon"));
+    let (child, port) = common::spawn_listening(|port| {
+        let tmp_dir = std::env::temp_dir().join(format!("moon-protoerr-{port}"));
+        let _ = std::fs::create_dir_all(&tmp_dir);
+        Command::new(&bin)
+            .args([
+                "--port",
+                &port.to_string(),
+                "--shards",
+                shards,
+                "--admin-port",
+                "0",
+                "--appendonly",
+                "no",
+                // This host hovers near the 5% diskfull line; the guard would
+                // turn writes into MOONERR and flake the suite.
+                "--disk-free-min-pct",
+                "0",
+                "--dir",
+                tmp_dir.to_str().unwrap(),
+            ])
+            .stdout(Stdio::null())
+            .stderr(
+                std::fs::File::create(tmp_dir.join("moon.stderr")).expect("create moon stderr log"),
+            )
+            .spawn()
+            .expect("spawn moon")
+    });
+    let tmp_dir = std::env::temp_dir().join(format!("moon-protoerr-{port}"));
+    let mut moon = Moon {
+        child,
+        port,
+        tmp_dir,
+    };
+    let deadline = Instant::now() + Duration::from_secs(30);
+    while Instant::now() < deadline {
+        if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) {
+            let _ = c.set_read_timeout(Some(Duration::from_millis(500)));
+            if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() {
+                let mut buf = [0u8; 64];
+                if let Ok(n) = c.read(&mut buf)
+                    && n > 0
+                    && buf.starts_with(b"+PONG")
+                {
+                    return moon;
+                }
+            }
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    // Never skip. A silent early return would let this suite report green
+    // while exercising no server at all.
+    let status = match moon.child.try_wait() {
+        Ok(Some(s)) => format!("exited with {s}"),
+        Ok(None) => "still running but never answered PING".to_string(),
+        Err(e) => format!("status unavailable: {e}"),
+    };
+    let log = std::fs::read_to_string(moon.tmp_dir.join("moon.stderr")).unwrap_or_default();
+    panic!("moon never became ready on port {port} ({status})\n--- stderr ---\n{log}");
+}
+
+/// What a server did with one payload: what it said, and what it did to the
+/// socket afterwards.
+#[derive(Debug)]
+struct Outcome {
+    reply: Vec<u8>,
+    closed: bool,
+    /// Reply to a follow-up PING, or `None` if the link was already gone.
+    /// `Some(b"")` means the connection is open but MUTE — a stall, which is
+    /// its own distinct failure and must never be mistaken for a clean close.
+    after: Option<Vec<u8>>,
+}
+
+impl Outcome {
+    fn reply_str(&self) -> String {
+        String::from_utf8_lossy(&self.reply).into_owned()
+    }
+    fn alive_and_serving(&self) -> bool {
+        matches!(&self.after, Some(a) if a.starts_with(b"+PONG"))
+    }
+}
+
+/// Send `payload`, drain the answer, then test whether the link survives.
+fn probe(port: u16, payload: &[u8]) -> Outcome {
+    let mut s = TcpStream::connect(("127.0.0.1", port)).expect("connect");
+    // Generous: a loaded CI host has truncated real replies into the
+    // follow-up read at 2s, which manufactures phantom "empty reply" results.
+    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    s.set_write_timeout(Some(Duration::from_secs(5))).unwrap();
+    s.write_all(payload).expect("write payload");
+
+    let mut reply = Vec::new();
+    let mut closed = false;
+    let mut buf = [0u8; 8192];
+    loop {
+        match s.read(&mut buf) {
+            Ok(0) => {
+                closed = true;
+                break;
+            }
+            Ok(n) => {
+                reply.extend_from_slice(&buf[..n]);
+                if reply.len() > 4096 {
+                    break;
+                }
+                // Keep draining briefly: an error frame and the close often
+                // arrive in separate segments, and stopping at the first read
+                // would report "did not close" for a server that did.
+                // NOT unwrapped: once the peer closes (which is exactly what
+                // an oversized request provokes), macOS fails this setsockopt
+                // with EINVAL. Panicking here would report a test bug as a
+                // server bug — pe4 did precisely that.
+                let _ = s.set_read_timeout(Some(Duration::from_millis(300)));
+            }
+            // A timeout means "nothing more is coming", NOT that the peer
+            // hung up. Conflating the two is how a stall gets misreported.
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
+            Err(e) if e.kind() == std::io::ErrorKind::TimedOut => break,
+            Err(_) => {
+                closed = true;
+                break;
+            }
+        }
+    }
+
+    if closed {
+        return Outcome {
+            reply,
+            closed,
+            after: None,
+        };
+    }
+
+    let _ = s.set_read_timeout(Some(Duration::from_secs(3)));
+    let after = match s.write_all(b"*1\r\n$4\r\nPING\r\n") {
+        Ok(()) => {
+            let mut a = Vec::new();
+            match s.read(&mut buf) {
+                Ok(0) => None,
+                Ok(n) => {
+                    a.extend_from_slice(&buf[..n]);
+                    Some(a)
+                }
+                // Open but mute: the stall case. Distinct from a close.
+                Err(_) => Some(Vec::new()),
+            }
+        }
+        Err(_) => None,
+    };
+    Outcome {
+        reply,
+        closed,
+        after,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// pe1 — the headline: a fault must not eat the valid frames before it.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe1_valid_prefix_is_answered_before_the_fault() {
+    let m = spawn_moon("1");
+    // ONE write, so both frames land in the same read buffer and the parse
+    // loop sees the valid PING and the bad frame together. That is precisely
+    // the case `Err(_) => break` drops on the floor: `batch` already holds
+    // the parsed PING and is discarded unexecuted.
+    let out = probe(m.port, b"*1\r\n$4\r\nPING\r\n*-9\r\n");
+    assert!(
+        out.reply.starts_with(b"+PONG"),
+        "the PING that arrived BEFORE the malformed frame must still be \
+         answered; redis-server 8.6.1 answers it. got reply={:?} closed={}",
+        out.reply_str(),
+        out.closed
+    );
+}
+
+// ---------------------------------------------------------------------------
+// pe2 — name the fault instead of a bare close.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe2_bad_bulk_length_names_itself_then_closes() {
+    let m = spawn_moon("1");
+    // Three shapes of the same fault: non-numeric, negative-but-not-(-1),
+    // and past the configured maximum. Redis answers all three identically.
+    for bad in [
+        &b"*2\r\n$3\r\nGET\r\n$abc\r\nk\r\n"[..],
+        &b"*2\r\n$3\r\nGET\r\n$-5\r\nk\r\n"[..],
+        &b"*2\r\n$3\r\nGET\r\n$999999999\r\nk\r\n"[..],
+    ] {
+        let out = probe(m.port, bad);
+        let reply = out.reply_str();
+        assert!(
+            reply.contains(ERR_BULK_LEN),
+            "malformed bulk length must be NAMED on the wire before the close.\n\
+             sent    : {:?}\n\
+             expected: contains {ERR_BULK_LEN:?}\n\
+             got     : {reply:?} (closed={})",
+            String::from_utf8_lossy(bad),
+            out.closed
+        );
+        assert!(
+            out.closed || out.after.is_none(),
+            "after a protocol fault the connection must close; sent {:?}",
+            String::from_utf8_lossy(bad)
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// pe3 — a negative multibulk count is IGNORED, not fatal. Measured Redis.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe3_negative_multibulk_count_is_ignored_and_connection_survives() {
+    let m = spawn_moon("1");
+    let out = probe(m.port, b"*-9\r\n");
+    assert!(
+        !out.closed,
+        "redis-server treats a negative multibulk count as a null array: it \
+         consumes the bytes, replies nothing, and keeps serving. Moon closes."
+    );
+    assert!(
+        out.alive_and_serving(),
+        "connection must still serve after `*-9`; follow-up PING got {:?}",
+        out.after
+            .as_ref()
+            .map(|a| String::from_utf8_lossy(a).into_owned())
+    );
+}
+
+// ---------------------------------------------------------------------------
+// pe4 — Moon already BUILDS this string; it just never reaches the client.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe4_oversized_inline_request_names_itself() {
+    let m = spawn_moon("1");
+    // Moon's cap bites between 65_530 and 70_000 bytes (measured). 200 KB is
+    // comfortably past it on both servers.
+    let mut payload = Vec::from(&b"GET "[..]);
+    payload.extend(std::iter::repeat_n(b'x', 200_000));
+    payload.extend_from_slice(b"\r\n");
+    let out = probe(m.port, &payload);
+    assert!(
+        out.reply_str().contains(ERR_INLINE_TOO_BIG),
+        "src/protocol/inline.rs already constructs exactly this message — the \
+         handler discards it. expected contains {ERR_INLINE_TOO_BIG:?}, got {:?}",
+        out.reply_str()
+    );
+}
+
+// ---------------------------------------------------------------------------
+// pe5 — an unbalanced quote is a protocol fault, not a key name.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe5_unbalanced_quote_is_rejected_not_silently_accepted() {
+    let m = spawn_moon("1");
+    let out = probe(m.port, b"GET \"unclosed\r\n");
+    let reply = out.reply_str();
+    assert!(
+        !reply.starts_with("$-1"),
+        "Moon currently ACCEPTS the unterminated quote, treating it as part of \
+         the key, and answers a nil lookup. redis-server rejects the request. \
+         got {reply:?}"
+    );
+    assert!(
+        reply.contains(ERR_UNBALANCED),
+        "expected contains {ERR_UNBALANCED:?}, got {reply:?}"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// pe6 — an error must not leave the connection open-but-mute.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe6_error_reply_does_not_stall_the_connection() {
+    let m = spawn_moon("1");
+    // `@bogus` is not a RESP type byte, so both servers route it to the inline
+    // parser and answer "unknown command". Both then answer the PING. Redis
+    // keeps serving after that; Moon stops answering — the stall this asserts.
+    let out = probe(m.port, b"@bogus\r\n*1\r\n$4\r\nPING\r\n");
+    assert!(
+        !out.closed,
+        "an unknown command must not close the connection"
+    );
+    assert!(
+        out.reply_str().contains("PONG"),
+        "the PING after the bad command must be answered; got {:?}",
+        out.reply_str()
+    );
+    assert!(
+        out.alive_and_serving(),
+        "the connection must keep serving AFTER an error reply. \
+         `Some(\"\")` here means open-but-mute — a stall, not a close. got {:?}",
+        out.after
+            .as_ref()
+            .map(|a| String::from_utf8_lossy(a).into_owned())
+    );
+}
+
+// ---------------------------------------------------------------------------
+// pe7 — the handlers must agree. A fix in one is not a fix.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe7_all_shard_counts_agree_on_protocol_faults() {
+    // shards=1 and shards=4 exercise different handler paths (the inline fast
+    // path is only reachable when the key is shard-local). A protocol fault
+    // must look identical regardless.
+    let one = spawn_moon("1");
+    let four = spawn_moon("4");
+    let payloads: [&[u8]; 4] = [
+        b"*2\r\n$3\r\nGET\r\n$abc\r\nk\r\n",
+        b"*-9\r\n",
+        b"GET \"unclosed\r\n",
+        b"*1\r\n$4\r\nPING\r\n*-9\r\n",
+    ];
+    for p in payloads {
+        let a = probe(one.port, p);
+        let b = probe(four.port, p);
+        assert_eq!(
+            a.reply_str(),
+            b.reply_str(),
+            "shards=1 and shards=4 disagree on {:?}",
+            String::from_utf8_lossy(p)
+        );
+        assert_eq!(
+            a.closed,
+            b.closed,
+            "shards=1 and shards=4 disagree on whether to close for {:?}",
+            String::from_utf8_lossy(p)
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// pe8 — a pure unit test over the wire text, so a typo fails without a server.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn pe8_wire_text_is_redis_verbatim() {
+    use moon::protocol::ProtoFault;
+    // Copied from redis 8.6.1's networking.c. If one of these ever needs to
+    // change, it is because Redis changed — not because Moon found it awkward.
+    assert_eq!(
+        ProtoFault::BulkLen.wire_text(),
+        "Protocol error: invalid bulk length"
+    );
+    assert_eq!(
+        ProtoFault::MultibulkLen.wire_text(),
+        "Protocol error: invalid multibulk length"
+    );
+    assert_eq!(
+        ProtoFault::InlineTooBig.wire_text(),
+        "Protocol error: too big inline request"
+    );
+    assert_eq!(
+        ProtoFault::UnbalancedQuotes.wire_text(),
+        "Protocol error: unbalanced quotes in request"
+    );
+    assert_eq!(
+        ProtoFault::MbulkCountTooBig.wire_text(),
+        "Protocol error: too big mbulk count string"
+    );
+    // The two byte-carrying variants format their offender in.
+    assert_eq!(
+        ProtoFault::ExpectedDollar(b',').wire_text_owned(),
+        "Protocol error: expected '$', got ','"
+    );
+    assert_eq!(
+        ProtoFault::UnknownType(b'@').wire_text_owned(),
+        "Protocol error: expected '$', got '@'"
+    );
+}