diff --git a/README.md b/README.md
index 5466998..bf26be0 100644
--- a/README.md
+++ b/README.md
@@ -404,7 +404,7 @@ The node-redis adapter owns no additional resources, so the application closes t
 
 Reads use native `GET` for untracked entries and one atomic `MGET` for each tracked value-and-watermark pair. The adapters validate and decode the returned frame in the Node process. Tracked reads are deliberately routed to primaries so a lagging replica cannot hide an invalidation watermark.
 
-Native commands retain Redis's wrong-type behavior. An untracked `GET` surfaces `WRONGTYPE`; tracked `MGET` represents a wrong-type member as a missing value. A wrong-type tracked value is therefore a clean miss and may be replaced with a valid DialCache frame after the fallback succeeds, while a wrong-type watermark prevents the tracked write from succeeding.
+Native commands retain Redis's wrong-type behavior. An untracked `GET` surfaces `WRONGTYPE`; tracked `MGET` represents a wrong-type member as a missing value. A wrong-type tracked value is therefore a clean `not_found` miss and may be replaced with a valid DialCache frame after the fallback succeeds, while a wrong-type watermark misses as `watermark_unreadable` and prevents the tracked write from succeeding.
 
 Node-redis forces tracked cluster commands to the slot primary. GLIDE uses an explicit primary route in cluster mode; in standalone mode it sends `MGET` through a one-command non-atomic batch because direct read commands follow the client's replica-read preference. Standalone batches use the primary, and `MGET` itself provides the atomic snapshot without consuming caller-owned `WATCH` state. The GLIDE helper distinguishes those modes from the direct client's runtime identity and rejects ambiguous clients instead of silently choosing a route.
 
@@ -426,7 +426,7 @@ Writes, invalidations, async `cacheConfigProvider` calls, and custom serializer
 
 #### Serialization
 
-The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. It exchanges serialized values as `string | Buffer` and does not expose client commands or wire encodings. The shared `decodeRedisFrame` and `decodeTrackedRedisFrame` helpers, write and invalidation Lua sources, and wire constants are available from `dialcache/redis-protocol`, so custom adapters can reuse the bundled adapters' exact miss and watermark-fencing rules. Custom adapters can throw the root-exported `DialCacheRedisPayloadError`, `DialCacheRedisPayloadEncodingError`, and `DialCacheRedisProtocolError` classes to distinguish malformed replies, unsupported encodings, and mutation-script reply-domain violations in logs. DialCache records bounded `cache_read`, `cache_write`, or `invalidation` metrics by failure site.
+The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. It exchanges serialized values as `string | Buffer` and does not expose client commands or wire encodings. `read()` returns a `RedisReadOutcome`: a hit carrying the decoded payload, or a miss carrying a bounded `RedisReadMissReason` (see [Miss reasons](#miss-reasons)). The shared `decodeRedisFrame` and `decodeTrackedRedisFrame` helpers produce that outcome directly; they, the write and invalidation Lua sources, and wire constants are available from `dialcache/redis-protocol`, so custom adapters can reuse the bundled adapters' exact miss-classification and watermark-fencing rules. DialCache validates each read outcome at runtime and treats a malformed one — including a legacy `payload | null` return — as a fail-open `cache_read` error rather than a miss, so miss reasons stay bounded. Custom adapters can throw the root-exported `DialCacheRedisPayloadError`, `DialCacheRedisPayloadEncodingError`, and `DialCacheRedisProtocolError` classes to distinguish malformed replies, unsupported encodings, and mutation-script reply-domain violations in logs. DialCache records bounded `cache_read`, `cache_write`, or `invalidation` metrics by failure site.
 
 Redis values use a compact binary frame:
 
@@ -556,7 +556,7 @@ The detached job uses this bounded algorithm:
 6. If `C1` is missing or differs byte-for-byte from `C0`, emit `superseded`; if it is identical, emit `mismatch`.
 7. If the confirmation read fails or reaches its Redis-read deadline, emit `confirmation_error`.
 
-Here a clean miss means the semantic Redis read returned `null`; it does not include a non-null payload that later fails deserialization. A caller fallback rejection or timeout never becomes accepted `S` and never starts the fill.
+Here a clean miss means the semantic Redis read returned a miss outcome (any `RedisReadMissReason`, including `watermark_invalidated`); it does not include a hit whose payload later fails deserialization. A caller fallback rejection or timeout never becomes accepted `S` and never starts the fill.
 
 Both detached Redis reads use the effective `remoteReadTimeoutMs` and the key's normal protocol. Every clean-miss fill uses the same serializer, TTL, and Redis-time timestamp as an ordinary fill. Tracked `C0` and `C1` reads remain watermark-aware and are routed to primaries by the bundled adapters, while tracked fills also retain the ordinary invalidation watermark. Untracked reads use the ordinary one-key read route, which has no shadow-specific primary guarantee, and untracked fills use the ordinary TTL write without a watermark. Strings compare exactly, Buffers compare by bytes, and string/Buffer pairs compare by their UTF-8 bytes. DialCache does not deserialize `C1`, compare it with `S`, or chase another version.
 
@@ -653,7 +653,7 @@ Invalidation writes a Redis watermark at `{encodedNamespace:encodedKeyType:encod
 
 The internal `:dialcache-frame-v1` suffix identifies values written with DialCache's binary protocol. Watermarks are stored as decimal timestamps.
 
-A cached Redis value whose Redis-created timestamp is older than or equal to the watermark is treated as stale and refreshed through fallback. `invalidateRemote(keyType, id, futureBufferMs)` sets the watermark to the greater of its existing value and Redis's current time plus the buffer. While that future window is active, an invocation that reaches the tracked Redis read treats the covered value as a miss. Native `MGET` must transfer an existing stale frame before the Node decoder can reject it, so completed reads can repeatedly pay the full stale-payload transfer during a nonzero buffer window. If a successful fallback then reaches the tracked Redis write while the watermark still fences it, Redis rejects the write, atomically unlinks that logically stale value key, and DialCache suppresses the corresponding process-local population; later reads of that entry avoid retransferring its payload. The fallback value still returns to its caller. A read failure or timeout never reaches that write-side cleanup, so a large stale value can continue to consume network bandwidth and trigger `cache_read_timeout` until another completed read cleans it up or its TTL expires. Request-local memoization remains unconditional. A ramped-out invocation without shadow work does not consult the watermark; a selected shadow path for that tracked key does consult it for `C0`, `C1` when needed, and any clean-miss fill, although caller-path request-local/process-local publication remains independent.
+A cached Redis value whose Redis-created timestamp is older than or equal to the watermark is treated as stale and refreshed through fallback. Fenced reads are directly observable as `miss` metrics with `reason="watermark_invalidated"`, separating invalidation churn from cold-key `not_found` misses (see [Miss reasons](#miss-reasons)). `invalidateRemote(keyType, id, futureBufferMs)` sets the watermark to the greater of its existing value and Redis's current time plus the buffer. While that future window is active, an invocation that reaches the tracked Redis read treats the covered value as a miss. Native `MGET` must transfer an existing stale frame before the Node decoder can reject it, so completed reads can repeatedly pay the full stale-payload transfer during a nonzero buffer window. If a successful fallback then reaches the tracked Redis write while the watermark still fences it, Redis rejects the write, atomically unlinks that logically stale value key, and DialCache suppresses the corresponding process-local population; later reads of that entry avoid retransferring its payload. The fallback value still returns to its caller. A read failure or timeout never reaches that write-side cleanup, so a large stale value can continue to consume network bandwidth and trigger `cache_read_timeout` until another completed read cleans it up or its TTL expires. Request-local memoization remains unconditional. A ramped-out invocation without shadow work does not consult the watermark; a selected shadow path for that tracked key does consult it for `C0`, `C1` when needed, and any clean-miss fill, although caller-path request-local/process-local publication remains independent.
 
 The bundled timestamp protocol assumes that system clocks are synchronized across every Redis node eligible for primary promotion. Redis does not guarantee that `TIME` is monotonic across nodes, and DialCache does not detect or compensate for cross-node clock skew. If this deployment assumption is violated, failover can temporarily suppress tracked cache fills or allow a pre-invalidation value to remain readable until it expires or a later invalidation advances the watermark past its timestamp.
 
@@ -783,7 +783,7 @@ The Prometheus adapter emits:
 | Metric | Type | Labels | Description |
 | --- | --- | --- | --- |
 | `dialcache_request_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache-layer requests that reached an enabled layer |
-| `dialcache_miss_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache misses |
+| `dialcache_miss_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer`, `reason` | Cache misses classified by a bounded reason (see [Miss reasons](#miss-reasons)) |
 | `dialcache_disabled_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer`, `reason` | Cache skips (`context`, `policy_disabled`, `invalid_ttl`, `invalid_ramp`, `ramped_down`, `config_error`) |
 | `dialcache_error_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer`, `error`, `in_fallback` | Cache/fallback errors classified by a bounded failure site |
 | `dialcache_invalidation_counter` | Counter | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched |
@@ -845,7 +845,7 @@ The Datadog adapter emits exact increments of `1` for counters and preserves sec
 | Metric | Type | Tags | Description |
 | --- | --- | --- | --- |
 | `dialcache.request.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache-layer requests that reached an enabled layer |
-| `dialcache.miss.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache misses |
+| `dialcache.miss.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer`, `reason` | Cache misses by bounded reason |
 | `dialcache.disabled.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer`, `reason` | Cache skips by bounded reason |
 | `dialcache.error.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer`, `error`, `in_fallback` | Cache/fallback errors by bounded failure site |
 | `dialcache.invalidation.count` | Count | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched |
@@ -862,6 +862,22 @@ The Datadog adapter emits exact increments of `1` for counters and preserves sec
 
 Observer throws and rejections from returned promises or thenables are isolated by DialCache's fail-open metrics boundary. Buffered transport failures that are not represented by a returned thenable happen outside that boundary, so configure the DogStatsD client's error handling and shutdown behavior as part of application ownership.
 
+### Miss reasons
+
+The `reason` label on the miss metric separates invalidation-driven misses from cold keys:
+
+| `reason` | Meaning |
+| --- | --- |
+| `not_found` | The key is absent or expired (also a wrong-type value under tracked `MGET`) |
+| `frame_unsupported` | A Redis value exists but is shorter than the frame header or has an unsupported frame version |
+| `watermark_unreadable` | A tracked read's watermark is missing, malformed, or not finite |
+| `watermark_invalidated` | A tracked frame was fenced because its Redis-created timestamp is at or before the watermark |
+| `deserialization_failed` | The Redis payload was read but `serializer.load` failed (paired with an `error="serialization_load"` event; when accompanied by a `fallback_raw` or `read_over_limit` compression outcome, the stored compression envelope was unreadable rather than the serializer drifting) |
+
+`request_local` and `local` layers have no frames or watermarks, so their misses are always `not_found` (an expired process-local entry is indistinguishable from an absent one). Only `remote` and `remote_shadow` reads produce the other reasons (`deserialization_failed` occurs only on `remote`), and only tracked keys can produce the watermark reasons. `watermark_invalidated` measures invalidation churn directly: sustained volume during a future-buffer window is the repeated stale-frame transfer cost described in [Targeted invalidation](#targeted-invalidation-and-watermarks). Because a completed fenced write unlinks the stale frame, a window typically labels only the first fenced read of an entry `watermark_invalidated`; later reads in that window miss as `not_found` until a write succeeds. `frame_unsupported` and `watermark_unreadable` should be near zero in steady state; sustained volume indicates external key corruption, protocol mixing, or watermark loss.
+
+These values are defined by the backend-neutral core and are identical for every metrics adapter.
+
 ### Error categories
 
 The `error` label reports where an operation failed rather than copying the thrown value's class or `Error.name`:
diff --git a/scripts/benchmark-request-local.mjs b/scripts/benchmark-request-local.mjs
index 8a0c3b9..c03018f 100644
--- a/scripts/benchmark-request-local.mjs
+++ b/scripts/benchmark-request-local.mjs
@@ -248,7 +248,7 @@ async function benchmarkRedisReadDeadlineCoalescing(fanout) {
       redisReadCalls += 1;
       started.resolve();
       await gate.promise;
-      return JSON.stringify("shared");
+      return { status: "hit", payload: JSON.stringify("shared") };
     },
     async write() {
       return true;
@@ -322,7 +322,7 @@ async function benchmarkSequentialTrackedRedisHits(iterations, { scenario, useCa
     async read({ watermarkKey }) {
       assert.equal(typeof watermarkKey, "string", "the benchmark must exercise tracked Redis reads");
       redisReadCalls += 1;
-      return JSON.stringify("shared");
+      return { status: "hit", payload: JSON.stringify("shared") };
     },
     async write() {
       redisWriteCalls += 1;
@@ -451,7 +451,7 @@ async function benchmarkDarkShadowDetachment() {
   assert.equal(redisReadCalls, 1, "the detached C0 read should have started");
   assert.equal(fallbackCalls, 1, "the caller and shadow validation must share one SoT invocation");
 
-  readGate.resolve(JSON.stringify(cachedValue));
+  readGate.resolve({ status: "hit", payload: JSON.stringify(cachedValue) });
   await nextTurn();
   assert.equal(await outcomeGate.promise, "mismatch");
   assert.equal(redisReadCalls, 2, "only a mismatch candidate should add confirmation C1");
@@ -479,7 +479,7 @@ async function benchmarkDarkShadowFillDetachment() {
     async read({ watermarkKey }) {
       assert.equal(typeof watermarkKey, "string", "dark shadow reads must remain tracked");
       redisReadCalls += 1;
-      return null;
+      return { status: "miss", reason: "not_found" };
     },
     async write({ watermarkKey }) {
       assert.equal(typeof watermarkKey, "string", "dark shadow fills must remain tracked");
diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs
index ac9620b..4b4e713 100644
--- a/scripts/test-package.mjs
+++ b/scripts/test-package.mjs
@@ -39,10 +39,14 @@ const rootConsumer = `import {
   type InvalidationMetricLabels,
   type MetricErrorKind,
   type MetricLayer,
+  type MissMetricLabels,
+  type MissReason,
   type ProcessCoalescingState,
   type RedisConfig,
   type RedisInvalidationRequest,
   type RedisReadContext,
+  type RedisReadMissReason,
+  type RedisReadOutcome,
   type RedisWriteRequest,
   type Serializer,
   type ShadowComparator,
@@ -148,8 +152,8 @@ const redisProtocolError = new DialCacheRedisProtocolError("Invalid DialCache Re
 const emptyRedisFrame = Buffer.alloc(10);
 emptyRedisFrame[0] = 1;
 emptyRedisFrame.writeBigUInt64BE(1n, 1);
-const decodedEmptyRedisPayload: string | Buffer | null = decodeRedisFrame(emptyRedisFrame);
-const decodedStaleRedisPayload: string | Buffer | null = decodeTrackedRedisFrame(
+const decodedEmptyRedisPayload: RedisReadOutcome = decodeRedisFrame(emptyRedisFrame);
+const decodedStaleRedisPayload: RedisReadOutcome = decodeTrackedRedisFrame(
   emptyRedisFrame,
   Buffer.from("1"),
 );
@@ -321,6 +325,26 @@ const disabledReasons: Readonly<Record<DisabledReason, true>> = {
 };
 // @ts-expect-error Missing configuration now means the documented disabled policy, not a separate reason.
 const legacyMissingConfigReason: DisabledReason = "missing_config";
+const missReasons: Readonly<Record<MissReason, true>> = {
+  not_found: true,
+  frame_unsupported: true,
+  watermark_unreadable: true,
+  watermark_invalidated: true,
+  deserialization_failed: true,
+};
+// @ts-expect-error Miss reasons are a bounded metric vocabulary, not free-form diagnostics.
+const unboundedMissReason: MissReason = "Tenant123Miss";
+// Redis clients can only produce the decoder subset; deserialization_failed is metrics-level.
+const decoderMissReason: RedisReadMissReason = "watermark_invalidated";
+// @ts-expect-error deserialization_failed is core-owned; clients cannot supply it.
+const forgedDecoderMissReason: RedisReadMissReason = "deserialization_failed";
+const missMetricLabels: MissMetricLabels = {
+  cacheNamespace: "consumer-cache",
+  useCase: "Load",
+  keyType: "id",
+  layer: CacheLayer.REMOTE,
+  reason: decoderMissReason,
+};
 const metricErrorKinds: Readonly<Record<MetricErrorKind, true>> = {
   key_construction: true,
   config_resolution: true,
@@ -365,7 +389,7 @@ const unboundedCompressionOutcome: CompressionOutcome = "inflated";
 
 const customRedisClient: DialCacheRedisClient = {
   // The optional second read argument preserves one-argument custom clients.
-  read: async () => Buffer.from([0, 255]),
+  read: async () => ({ status: "hit", payload: Buffer.from([0, 255]) }),
   write: async ({ value }) => typeof value === "string" || Buffer.isBuffer(value),
   invalidate: async () => undefined,
 };
@@ -480,6 +504,10 @@ void requestLocalCoalescingScope;
 void boundedErrorKind;
 void disabledReasons;
 void legacyMissingConfigReason;
+void missReasons;
+void unboundedMissReason;
+void forgedDecoderMissReason;
+void missMetricLabels;
 void MissingKeyConfigError;
 void disabledOverlay;
 void metricErrorKinds;
@@ -725,11 +753,13 @@ if (
 const esmEmptyFrame = Buffer.alloc(10);
 esmEmptyFrame[0] = 1;
 esmEmptyFrame.writeBigUInt64BE(1n, 1);
-if (redisProtocol.decodeRedisFrame(esmEmptyFrame) !== "") {
+const esmEmptyOutcome = redisProtocol.decodeRedisFrame(esmEmptyFrame);
+if (esmEmptyOutcome.status !== "hit" || esmEmptyOutcome.payload !== "") {
   throw new Error("The packed ESM Redis protocol decoder did not preserve an empty UTF-8 payload");
 }
-if (redisProtocol.decodeTrackedRedisFrame(esmEmptyFrame, Buffer.from("1")) !== null) {
-  throw new Error("The packed ESM Redis protocol decoder did not reject a stale tracked frame");
+const esmStaleOutcome = redisProtocol.decodeTrackedRedisFrame(esmEmptyFrame, Buffer.from("1"));
+if (esmStaleOutcome.status !== "miss" || esmStaleOutcome.reason !== "watermark_invalidated") {
+  throw new Error("The packed ESM Redis protocol decoder did not fence a stale tracked frame");
 }
 try {
   redisProtocol.decodeRedisFrame("not binary");
@@ -860,7 +890,7 @@ console.log("${observerIsolationMarker}");`,
 let payload = Buffer.alloc(4 * 1024 * 1024, 1);
 const payloadReference = new WeakRef(payload);
 const redis = {
-  read: async () => payload,
+  read: async () => ({ status: "hit", payload }),
   write: async () => true,
   invalidate: async () => undefined,
 };
@@ -1003,11 +1033,13 @@ if (
 const cjsEmptyFrame = Buffer.alloc(10);
 cjsEmptyFrame[0] = 1;
 cjsEmptyFrame.writeBigUInt64BE(1n, 1);
-if (redisProtocol.decodeRedisFrame(cjsEmptyFrame) !== "") {
+const cjsEmptyOutcome = redisProtocol.decodeRedisFrame(cjsEmptyFrame);
+if (cjsEmptyOutcome.status !== "hit" || cjsEmptyOutcome.payload !== "") {
   throw new Error("The packed CommonJS Redis protocol decoder did not preserve an empty UTF-8 payload");
 }
-if (redisProtocol.decodeTrackedRedisFrame(cjsEmptyFrame, Buffer.from("1")) !== null) {
-  throw new Error("The packed CommonJS Redis protocol decoder did not reject a stale tracked frame");
+const cjsStaleOutcome = redisProtocol.decodeTrackedRedisFrame(cjsEmptyFrame, Buffer.from("1"));
+if (cjsStaleOutcome.status !== "miss" || cjsStaleOutcome.reason !== "watermark_invalidated") {
+  throw new Error("The packed CommonJS Redis protocol decoder did not fence a stale tracked frame");
 }
 try {
   redisProtocol.decodeRedisFrame("not binary");
diff --git a/src/datadog.ts b/src/datadog.ts
index d3e7dee..82833c9 100644
--- a/src/datadog.ts
+++ b/src/datadog.ts
@@ -7,6 +7,7 @@ import type {
   DisabledMetricLabels,
   ErrorMetricLabels,
   InvalidationMetricLabels,
+  MissMetricLabels,
   SerializationMetricLabels,
   ShadowValidationMetricLabels,
 } from "./metrics.js";
@@ -86,8 +87,8 @@ export class DatadogDialCacheMetrics implements DialCacheMetricsAdapter {
     this.increment(this.metricNames.request, cacheTags(labels));
   }
 
-  miss(labels: CacheMetricLabels): void {
-    this.increment(this.metricNames.miss, cacheTags(labels));
+  miss(labels: MissMetricLabels): void {
+    this.increment(this.metricNames.miss, { ...cacheTags(labels), reason: labels.reason });
   }
 
   disabled(labels: DisabledMetricLabels): void {
diff --git a/src/dialcache.ts b/src/dialcache.ts
index 62954b4..d72390e 100644
--- a/src/dialcache.ts
+++ b/src/dialcache.ts
@@ -564,7 +564,7 @@ export class DialCache {
         return result.value;
       }
 
-      this.metrics?.miss(labelsFor(key, REQUEST_LOCAL_CACHE_LAYER));
+      this.metrics?.miss({ ...labelsFor(key, REQUEST_LOCAL_CACHE_LAYER), reason: "not_found" });
       const value = await this.getThroughSharedLayers(
         key,
         keyConfig,
@@ -1183,7 +1183,7 @@ export class DialCache {
       this.metrics?.request(labelsFor(key, CacheLayer.LOCAL));
       this.metrics?.observeGet(labelsFor(key, CacheLayer.LOCAL), elapsedSeconds(start));
       if (result.status === "miss") {
-        this.metrics?.miss(labelsFor(key, CacheLayer.LOCAL));
+        this.metrics?.miss({ ...labelsFor(key, CacheLayer.LOCAL), reason: "not_found" });
       }
       return result;
     } catch (error) {
diff --git a/src/index.ts b/src/index.ts
index 116f80c..0c2bb14 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -15,6 +15,8 @@ export type {
   InvalidationMetricLabels,
   MetricErrorKind,
   MetricLayer,
+  MissMetricLabels,
+  MissReason,
   SerializationMetricLabels,
   ShadowValidationMetricLabels,
   ShadowValidationOutcome,
@@ -51,6 +53,8 @@ export type {
   RedisCachePayload,
   RedisInvalidationRequest,
   RedisReadContext,
+  RedisReadMissReason,
+  RedisReadOutcome,
   RedisReadRequest,
   RedisWriteRequest,
 } from "./redis-client.js";
diff --git a/src/internal/redis-cache.ts b/src/internal/redis-cache.ts
index 41eabed..8b17be4 100644
--- a/src/internal/redis-cache.ts
+++ b/src/internal/redis-cache.ts
@@ -10,7 +10,13 @@ import {
   type MetricErrorKind,
   type MetricLayer,
 } from "../metrics.js";
-import type { DialCacheRedisClient, RedisCachePayload } from "../redis-client.js";
+import {
+  DialCacheRedisPayloadError,
+  type DialCacheRedisClient,
+  type RedisCachePayload,
+  type RedisReadMissReason,
+  type RedisReadOutcome,
+} from "../redis-client.js";
 import { JsonSerializer, type Serializer } from "../serializer.js";
 import type { RedisCacheGetResult } from "./cache-result.js";
 import {
@@ -21,6 +27,7 @@ import {
   type CompressionConfig,
 } from "./compression.js";
 import { assertValidDeadlineMs, withMonotonicDeadline } from "./deadline.js";
+import { REDIS_READ_MISS_OUTCOMES } from "./redis-payload.js";
 import { cacheTtlSecToMs } from "./duration.js";
 import { fetchKeyConfig, resolveLayerConfigResult, type ResolvedLayerConfig } from "./runtime-config.js";
 
@@ -51,9 +58,9 @@ interface RedisCacheOptions {
   readonly metrics: DialCacheMetricsAdapter | null;
 }
 
-interface StartedRedisRead {
+interface StartedRead<T> {
   /** Result bounded by the effective Redis read deadline. */
-  readonly result: Promise<RedisCachePayload | null>;
+  readonly result: Promise<T>;
   /** Fulfills only after the underlying semantic Redis read settles. */
   readonly settled: Promise<void>;
 }
@@ -62,6 +69,50 @@ const defaultSerializer = new JsonSerializer<unknown>();
 const REDIS_FRAME_KEY_SUFFIX = ":dialcache-frame-v1";
 const DEFAULT_REMOTE_READ_TIMEOUT_MS = 50;
 
+/**
+ * Validate a client read result and return its canonical form: misses collapse
+ * onto the shared frozen singletons and hits are recaptured, so everything
+ * downstream — including the bounded metric labels — observes only validated
+ * data even when a client returns accessor-backed or otherwise unstable
+ * objects. Each client-owned property is read exactly once. Rejections (legacy
+ * `payload | null` results, unknown shapes, the forged core-owned
+ * `deserialization_failed` reason) throw with a bounded shape fingerprint and
+ * fail open through the existing cache_read error path.
+ */
+function normalizeRedisReadOutcome(outcome: RedisReadOutcome): RedisReadOutcome {
+  const result: unknown = outcome;
+  if (result === null || typeof result === "string" || Buffer.isBuffer(result)) {
+    throw invalidReadOutcome("a legacy `payload | null` result; return a RedisReadOutcome");
+  }
+  if (typeof result !== "object") {
+    throw invalidReadOutcome(`a value of type ${typeof result}`);
+  }
+  const { status, reason, payload } = result as { status?: unknown; reason?: unknown; payload?: unknown };
+  if (status === "hit") {
+    if (typeof payload === "string" || Buffer.isBuffer(payload)) {
+      return { status: "hit", payload };
+    }
+    throw invalidReadOutcome("a hit without a string or Buffer payload");
+  }
+  if (status !== "miss") {
+    throw invalidReadOutcome('a status other than "hit" or "miss"');
+  }
+  if (typeof reason === "string" && Object.hasOwn(REDIS_READ_MISS_OUTCOMES, reason)) {
+    return REDIS_READ_MISS_OUTCOMES[reason as RedisReadMissReason];
+  }
+  throw invalidReadOutcome(
+    reason === "deserialization_failed"
+      ? 'a miss with the core-owned reason "deserialization_failed"'
+      : "a miss without a bounded RedisReadMissReason",
+  );
+}
+
+function invalidReadOutcome(received: string): DialCacheRedisPayloadError {
+  return new DialCacheRedisPayloadError(
+    `Invalid DialCache Redis read outcome; expected { status: "hit" | "miss" } but received ${received}`,
+  );
+}
+
 export class RedisCache {
   private readonly configProvider: CacheConfigProvider;
   private readonly defaultSerializer: Serializer<unknown>;
@@ -126,9 +177,9 @@ export class RedisCache {
     const start = performance.now();
     this.recordMetric((metrics) => metrics.request(labelsFor(key, metricLayer)));
     try {
-      let payload: RedisCachePayload | null;
+      let outcome: RedisReadOutcome;
       try {
-        payload = await this.startPayloadRead(key, readTimeoutMs, false).result;
+        outcome = await this.startPayloadRead(key, readTimeoutMs, false).result;
       } catch (error) {
         this.recordError(
           key,
@@ -137,16 +188,16 @@ export class RedisCache {
         );
         throw error;
       }
-      if (payload === null) {
-        this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer)));
+      if (outcome.status === "miss") {
+        this.recordMetric((metrics) => metrics.miss({ ...labelsFor(key, metricLayer), reason: outcome.reason }));
         return { status: "miss", config: layerConfig };
       }
 
       try {
-        const value = await this.deserializePayload<T>(key, payload, metricLayer);
-        return { status: "hit", value, payload };
+        const value = await this.deserializePayload<T>(key, outcome.payload, metricLayer);
+        return { status: "hit", value, payload: outcome.payload };
       } catch {
-        this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer)));
+        this.recordMetric((metrics) => metrics.miss({ ...labelsFor(key, metricLayer), reason: "deserialization_failed" }));
         return { status: "miss", config: layerConfig };
       }
     } finally {
@@ -172,13 +223,19 @@ export class RedisCache {
   startPayloadReadForShadow(
     key: DialCacheKey,
     readTimeoutMs: number,
-  ): StartedRedisRead {
-    return this.startMeasuredPayloadRead(
+  ): StartedRead<RedisCachePayload | null> {
+    const read = this.startMeasuredPayloadRead(
       key,
       readTimeoutMs,
       REMOTE_SHADOW_CACHE_LAYER,
       true,
     );
+    // Shadow fill and confirmation only need presence; a fenced miss stays
+    // fill-eligible because the tracked write script re-fences server-side.
+    return {
+      result: read.result.then((outcome) => (outcome.status === "hit" ? outcome.payload : null)),
+      settled: read.settled,
+    };
   }
 
   async put<T>(key: DialCacheKey, value: T, config?: { readonly ttlSec: number }): Promise<boolean> {
@@ -308,7 +365,7 @@ export class RedisCache {
     key: DialCacheKey,
     readTimeoutMs: number,
     unrefTimer: boolean,
-  ): StartedRedisRead {
+  ): StartedRead<RedisReadOutcome> {
     const abortController = new AbortController();
     const pending = Promise.resolve().then(() =>
       this.client.read(
@@ -318,7 +375,7 @@ export class RedisCache {
         },
         { timeoutMs: readTimeoutMs, signal: abortController.signal },
       )
-    );
+    ).then(normalizeRedisReadOutcome);
     const result = withMonotonicDeadline({
       timeoutMs: readTimeoutMs,
       operation: () => pending,
@@ -340,16 +397,16 @@ export class RedisCache {
     readTimeoutMs: number,
     metricLayer: MetricLayer,
     unrefTimer: boolean,
-  ): StartedRedisRead {
+  ): StartedRead<RedisReadOutcome> {
     const start = performance.now();
     this.recordMetric((metrics) => metrics.request(labelsFor(key, metricLayer)));
     const read = this.startPayloadRead(key, readTimeoutMs, unrefTimer);
     const result = read.result.then(
-      (payload) => {
-        if (payload === null) {
-          this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer)));
+      (outcome) => {
+        if (outcome.status === "miss") {
+          this.recordMetric((metrics) => metrics.miss({ ...labelsFor(key, metricLayer), reason: outcome.reason }));
         }
-        return payload;
+        return outcome;
       },
       (error: unknown) => {
         this.recordError(
diff --git a/src/internal/redis-payload.ts b/src/internal/redis-payload.ts
index 1210678..5ab039e 100644
--- a/src/internal/redis-payload.ts
+++ b/src/internal/redis-payload.ts
@@ -2,6 +2,8 @@ import {
   DialCacheRedisPayloadEncodingError,
   DialCacheRedisPayloadError,
   type RedisCachePayload,
+  type RedisReadMissReason,
+  type RedisReadOutcome,
 } from "../redis-client.js";
 import {
   REDIS_ENCODING_BINARY,
@@ -21,10 +23,24 @@ function validateRedisBulkStringReply(raw: unknown): Buffer | null {
   );
 }
 
-function isSupportedRedisFrame(raw: Buffer | null): raw is Buffer {
-  return raw !== null
-    && raw.length >= REDIS_FRAME_MIN_BYTES
-    && raw[0] === REDIS_FRAME_VERSION;
+/**
+ * Canonical miss outcomes: one shared frozen instance per bounded read-miss
+ * reason. The decoders return these, and the core normalizes every client
+ * miss onto them, so a given reason has exactly one outcome object anywhere
+ * in the library. The Record annotation pins the table to the
+ * RedisReadMissReason vocabulary — never the metrics-level superset, which
+ * would let clients forge the core-owned deserialization_failed reason.
+ */
+export const REDIS_READ_MISS_OUTCOMES: Readonly<Record<RedisReadMissReason, RedisReadOutcome>> =
+  Object.freeze({
+    not_found: Object.freeze({ status: "miss", reason: "not_found" }),
+    frame_unsupported: Object.freeze({ status: "miss", reason: "frame_unsupported" }),
+    watermark_unreadable: Object.freeze({ status: "miss", reason: "watermark_unreadable" }),
+    watermark_invalidated: Object.freeze({ status: "miss", reason: "watermark_invalidated" }),
+  });
+
+function isSupportedRedisFrame(raw: Buffer): boolean {
+  return raw.length >= REDIS_FRAME_MIN_BYTES && raw[0] === REDIS_FRAME_VERSION;
 }
 
 function parseRedisWatermark(raw: Buffer | null): number | null {
@@ -58,37 +74,48 @@ function decodeRedisPayload(raw: Buffer): RedisCachePayload {
 
 /**
  * Decode an untracked DialCache frame returned as a Redis bulk string.
- * Missing, short, and unsupported-version frames are cache misses. Invalid
- * runtime reply types and unsupported payload encodings throw typed errors.
+ * A missing frame misses as `not_found`; a short or unsupported-version frame
+ * misses as `frame_unsupported`. Invalid runtime reply types and unsupported
+ * payload encodings throw typed errors.
  */
-export function decodeRedisFrame(raw: unknown): RedisCachePayload | null {
+export function decodeRedisFrame(raw: unknown): RedisReadOutcome {
   const frame = validateRedisBulkStringReply(raw);
-  return isSupportedRedisFrame(frame)
-    ? decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES))
-    : null;
+  if (frame === null) {
+    return REDIS_READ_MISS_OUTCOMES.not_found;
+  }
+  if (!isSupportedRedisFrame(frame)) {
+    return REDIS_READ_MISS_OUTCOMES.frame_unsupported;
+  }
+  return { status: "hit", payload: decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)) };
 }
 
 /**
  * Decode a tracked DialCache frame against a watermark from the same atomic,
- * authoritative snapshot. Missing or malformed state and frames created at or
- * before the watermark are cache misses. Invalid runtime reply types and
- * unsupported payload encodings throw typed errors.
+ * authoritative snapshot. Frame state is classified before watermark state:
+ * a missing frame misses as `not_found` and a short or unsupported-version
+ * frame as `frame_unsupported` regardless of the watermark. A missing or
+ * malformed watermark misses as `watermark_unreadable`, and a frame created
+ * at or before the watermark is fenced as `watermark_invalidated`. Invalid
+ * runtime reply types and unsupported payload encodings throw typed errors.
  */
 export function decodeTrackedRedisFrame(
   raw: unknown,
   rawWatermark: unknown,
-): RedisCachePayload | null {
+): RedisReadOutcome {
   const frame = validateRedisBulkStringReply(raw);
   const watermarkFrame = validateRedisBulkStringReply(rawWatermark);
+  if (frame === null) {
+    return REDIS_READ_MISS_OUTCOMES.not_found;
+  }
   if (!isSupportedRedisFrame(frame)) {
-    return null;
+    return REDIS_READ_MISS_OUTCOMES.frame_unsupported;
   }
   const watermark = parseRedisWatermark(watermarkFrame);
   if (watermark === null) {
-    return null;
+    return REDIS_READ_MISS_OUTCOMES.watermark_unreadable;
   }
   const createdAtMs = Number(frame.readBigUInt64BE(1));
   return createdAtMs <= watermark
-    ? null
-    : decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES));
+    ? REDIS_READ_MISS_OUTCOMES.watermark_invalidated
+    : { status: "hit", payload: decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)) };
 }
diff --git a/src/metrics.ts b/src/metrics.ts
index 58f8574..9a76698 100644
--- a/src/metrics.ts
+++ b/src/metrics.ts
@@ -1,5 +1,6 @@
 import type { CacheLayer } from "./config.js";
 import type { DialCacheKey } from "./key.js";
+import type { RedisReadMissReason } from "./redis-client.js";
 
 export const NO_CACHE_LAYER = "noop";
 export const REMOTE_SHADOW_CACHE_LAYER = "remote_shadow";
@@ -43,6 +44,8 @@ export type CompressionOutcome =
   | "read_over_limit";
 /** Bounded reasons for skipping cache work; policy_disabled means a shared layer has no effective TTL. */
 export type DisabledReason = "context" | "policy_disabled" | "invalid_ttl" | "invalid_ramp" | "ramped_down" | "config_error";
+/** Bounded reasons a cache read counted as a miss; local layers always use not_found. */
+export type MissReason = RedisReadMissReason | "deserialization_failed";
 /** Stable failure sites used instead of backend- or application-defined error names. */
 export type MetricErrorKind =
   | "key_construction"
@@ -65,6 +68,10 @@ export interface CacheMetricLabels {
   readonly layer: MetricLayer;
 }
 
+export interface MissMetricLabels extends CacheMetricLabels {
+  readonly reason: MissReason;
+}
+
 export interface DisabledMetricLabels extends CacheMetricLabels {
   readonly reason: DisabledReason;
 }
@@ -108,7 +115,7 @@ export interface ShadowValidationMetricLabels {
 
 export interface DialCacheMetricsAdapter {
   request(labels: CacheMetricLabels): void;
-  miss(labels: CacheMetricLabels): void;
+  miss(labels: MissMetricLabels): void;
   disabled(labels: DisabledMetricLabels): void;
   error(labels: ErrorMetricLabels): void;
   invalidation(labels: InvalidationMetricLabels): void;
diff --git a/src/prometheus.ts b/src/prometheus.ts
index c379776..416295f 100644
--- a/src/prometheus.ts
+++ b/src/prometheus.ts
@@ -9,6 +9,7 @@ import type {
   DisabledMetricLabels,
   ErrorMetricLabels,
   InvalidationMetricLabels,
+  MissMetricLabels,
   SerializationMetricLabels,
   ShadowValidationMetricLabels,
 } from "./metrics.js";
@@ -21,6 +22,7 @@ export interface PrometheusMetricsOptions {
 type PrometheusRegistry = Registry | Registry<OpenMetricsContentType>;
 
 type CounterLabels = "cache_namespace" | "use_case" | "key_type" | "layer";
+type MissLabels = CounterLabels | "reason";
 type DisabledLabels = CounterLabels | "reason";
 type ErrorLabels = CounterLabels | "error" | "in_fallback";
 type SerializationLabels = CounterLabels | "operation";
@@ -61,7 +63,7 @@ const RATIO_BUCKETS = [0.05, 0.1, 0.2, 0.3, 0.5, 0.7, 0.9, 1];
 
 export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter {
   private readonly requestCounter: Counter<CounterLabels>;
-  private readonly missCounter: Counter<CounterLabels>;
+  private readonly missCounter: Counter<MissLabels>;
   private readonly disabledCounter: Counter<DisabledLabels>;
   private readonly errorCounter: Counter<ErrorLabels>;
   private readonly invalidationCounter: Counter<InvalidationLabels>;
@@ -103,8 +105,8 @@ export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter {
     this.requestCounter.inc(cacheLabels(labels));
   }
 
-  miss(labels: CacheMetricLabels): void {
-    this.missCounter.inc(cacheLabels(labels));
+  miss(labels: MissMetricLabels): void {
+    this.missCounter.inc({ ...cacheLabels(labels), reason: labels.reason });
   }
 
   disabled(labels: DisabledMetricLabels): void {
@@ -203,7 +205,7 @@ function collectorConfigs(prefix: string) {
       type: "counter",
       name: `${prefix}dialcache_miss_counter`,
       help: "DialCache cache misses.",
-      labelNames: ["cache_namespace", "use_case", "key_type", "layer"],
+      labelNames: ["cache_namespace", "use_case", "key_type", "layer", "reason"],
     },
     requestCounter: {
       type: "counter",
diff --git a/src/redis-client.ts b/src/redis-client.ts
index aab67c8..11b199e 100644
--- a/src/redis-client.ts
+++ b/src/redis-client.ts
@@ -61,6 +61,22 @@ export class DialCacheRedisProtocolError extends Error {
 /** Serialized cache data, independent of any Redis client or wire framing. */
 export type RedisCachePayload = string | Buffer;
 
+/** Bounded classification for why a Redis read produced no payload. */
+export type RedisReadMissReason =
+  /** The value key is absent or expired, or a tracked MGET value member holds a non-string type. */
+  | "not_found"
+  /** The value frame is shorter than its header or has an unsupported version. */
+  | "frame_unsupported"
+  /** Tracked read whose watermark is missing, malformed, or not finite. */
+  | "watermark_unreadable"
+  /** Tracked read whose frame was created at or before the watermark. */
+  | "watermark_invalidated";
+
+/** Result of decoding one DialCache Redis read from an authoritative snapshot. */
+export type RedisReadOutcome =
+  | { readonly status: "hit"; readonly payload: RedisCachePayload }
+  | { readonly status: "miss"; readonly reason: RedisReadMissReason };
+
 interface RedisValueRequest {
   readonly valueKey: string;
 }
@@ -118,28 +134,31 @@ export interface RedisInvalidationRequest {
  */
 export interface DialCacheRedisClient {
   /**
-   * Read a DialCache Redis frame and return its decoded serializer payload.
+   * Read a DialCache Redis frame and return its decoded read outcome.
    * Implementations must use `decodeRedisFrame` / `decodeTrackedRedisFrame`
    * from `dialcache/redis-protocol`, or preserve their exact behavior.
    *
-   * Raw values are Redis bulk strings (`Buffer`) or null. A missing value, a
-   * frame shorter than the version/timestamp/encoding header, or an
-   * unsupported frame version is a cache miss. A tracked read also misses
-   * when its watermark is missing, is not a finite unsigned decimal, or is
-   * greater than or equal to the frame's creation time. In other words,
-   * `createdAt <= watermark` is fenced. Unsupported payload encodings and
-   * non-bulk runtime replies are payload protocol errors rather than misses.
+   * Raw values are Redis bulk strings (`Buffer`) or null. A miss carries one
+   * bounded `RedisReadMissReason`: a missing value is `not_found`; a frame
+   * shorter than the version/timestamp/encoding header or with an unsupported
+   * frame version is `frame_unsupported`; a tracked read whose watermark is
+   * missing or is not a finite unsigned decimal is `watermark_unreadable`;
+   * and a tracked frame with `createdAt <= watermark` is fenced as
+   * `watermark_invalidated`. Frame state is classified before watermark
+   * state, so a missing frame misses as `not_found` regardless of its
+   * watermark. Unsupported payload encodings and non-bulk runtime replies are
+   * payload protocol errors rather than misses.
    *
    * Tracked implementations must read the value and watermark atomically from
    * one authoritative snapshot; replica lag must not hide an invalidation.
    *
-   * A non-null payload is transferred to DialCache. A returned Buffer must
-   * remain stable and must not be mutated, pooled, or reused after this method
+   * A hit payload is transferred to DialCache. A returned Buffer must remain
+   * stable and must not be mutated, pooled, or reused after this method
    * settles; DialCache may retain it beyond the request for best-effort shadow
    * deserialization. Adapters that recycle response storage must return a
    * dedicated Buffer.
    */
-  read(request: RedisReadRequest, context?: RedisReadContext): Awaitable<RedisCachePayload | null>;
+  read(request: RedisReadRequest, context?: RedisReadContext): Awaitable<RedisReadOutcome>;
   /** Atomically write using server time. False means invalidation blocked the write. */
   write(request: RedisWriteRequest): Awaitable<boolean>;
   /**
diff --git a/src/redis-protocol.ts b/src/redis-protocol.ts
index 9606653..8f7508b 100644
--- a/src/redis-protocol.ts
+++ b/src/redis-protocol.ts
@@ -21,3 +21,4 @@ export {
   decodeRedisFrame,
   decodeTrackedRedisFrame,
 } from "./internal/redis-payload.js";
+export type { RedisReadMissReason, RedisReadOutcome } from "./redis-client.js";
diff --git a/test/datadog.test.ts b/test/datadog.test.ts
index cfd083c..3d697f8 100644
--- a/test/datadog.test.ts
+++ b/test/datadog.test.ts
@@ -7,6 +7,7 @@ import {
   DialCacheKeyConfig,
   type CompressionOutcome,
   type DisabledReason,
+  type MissReason,
   type DialCacheRedisClient,
   type MetricErrorKind,
   type MetricLayer,
@@ -90,6 +91,14 @@ const DISABLED_REASONS: Readonly<Record<DisabledReason, true>> = {
   config_error: true,
 };
 const disabledReasons = Object.keys(DISABLED_REASONS) as DisabledReason[];
+const MISS_REASONS: Readonly<Record<MissReason, true>> = {
+  not_found: true,
+  frame_unsupported: true,
+  watermark_unreadable: true,
+  watermark_invalidated: true,
+  deserialization_failed: true,
+};
+const missReasons = Object.keys(MISS_REASONS) as MissReason[];
 const ERROR_KINDS: Readonly<Record<MetricErrorKind, true>> = {
   key_construction: true,
   config_resolution: true,
@@ -144,7 +153,7 @@ describe("Datadog metrics adapter", () => {
     const metrics = new DatadogDialCacheMetrics({ client, observationMetricType: "distribution" });
 
     metrics.request(cacheLabels);
-    metrics.miss(cacheLabels);
+    metrics.miss({ ...cacheLabels, reason: "not_found" });
     metrics.disabled({ ...cacheLabels, reason: "ramped_down" });
     metrics.error({ ...cacheLabels, error: "cache_read", inFallback: true });
     metrics.invalidation({ cacheNamespace: cacheLabels.cacheNamespace, keyType: "user_id", layer: CacheLayer.REMOTE });
@@ -172,7 +181,7 @@ describe("Datadog metrics adapter", () => {
     const baseTags = { cache_namespace: "users", use_case: "LoadUser", key_type: "user_id", layer: "local" };
     expect(client.calls).toEqual([
       { method: "increment", name: "dialcache.request.count", value: 1, tags: baseTags },
-      { method: "increment", name: "dialcache.miss.count", value: 1, tags: baseTags },
+      { method: "increment", name: "dialcache.miss.count", value: 1, tags: { ...baseTags, reason: "not_found" } },
       {
         method: "increment",
         name: "dialcache.disabled.count",
@@ -270,6 +279,9 @@ describe("Datadog metrics adapter", () => {
     for (const reason of disabledReasons) {
       metrics.disabled({ ...cacheLabels, reason });
     }
+    for (const reason of missReasons) {
+      metrics.miss({ ...cacheLabels, reason });
+    }
     for (const error of errorKinds) {
       metrics.error({ ...cacheLabels, error, inFallback: false });
       metrics.error({ ...cacheLabels, error, inFallback: true });
@@ -303,6 +315,11 @@ describe("Datadog metrics adapter", () => {
         .filter(({ name }) => name === "dialcache.disabled.count")
         .map(({ tags }) => tags.reason),
     ).toEqual(disabledReasons);
+    expect(
+      client.calls
+        .filter(({ name }) => name === "dialcache.miss.count")
+        .map(({ tags }) => tags.reason),
+    ).toEqual(missReasons);
     expect(
       client.calls
         .filter(({ name }) => name === "dialcache.error.count")
diff --git a/test/dialcache-coalescing.test.ts b/test/dialcache-coalescing.test.ts
index f51c408..9e8889e 100644
--- a/test/dialcache-coalescing.test.ts
+++ b/test/dialcache-coalescing.test.ts
@@ -250,6 +250,7 @@ describe("DialCache request coalescing", () => {
       useCase: "RequestThenProcessCoalescing",
       keyType: "user_id",
       layer: "request_local",
+      reason: "not_found",
     });
     expect(miss.mock.calls.filter(([labels]) => labels.layer === "request_local")).toHaveLength(2);
   });
diff --git a/test/dialcache-invalidation.test.ts b/test/dialcache-invalidation.test.ts
index 9bdb024..20f9a74 100644
--- a/test/dialcache-invalidation.test.ts
+++ b/test/dialcache-invalidation.test.ts
@@ -261,7 +261,8 @@ describe("DialCache targeted invalidation watermarks", () => {
   it("treats a tracked value with a missing watermark marker as a miss", async () => {
     const redis = new FakeRedis();
     redis.setRaw(valueKey("MissingWatermark"), encodeFrame({ source: "stale" }));
-    const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } });
+    const metrics = new RecordingMetrics();
+    const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, metrics });
     let calls = 0;
     const getUser = dialcache.cached(async (userId: string) => ({ userId, source: `fallback-${++calls}` }), {
       keyType: "user_id",
@@ -277,6 +278,55 @@ describe("DialCache targeted invalidation watermarks", () => {
     expect(first).toEqual({ userId: "123", source: "fallback-1" });
     expect(second).toEqual({ userId: "123", source: "fallback-1" });
     expect(redis.readWatermarkValue(watermarkKey)).toBe(0);
+    expect(metrics.events).toContainEqual({
+      name: "miss",
+      labels: {
+        cacheNamespace: "urn",
+        useCase: "MissingWatermark",
+        keyType: "user_id",
+        layer: CacheLayer.REMOTE,
+        reason: "watermark_unreadable",
+      },
+    });
+  });
+
+  it("classifies remote miss reasons for cold keys and watermark-fenced frames", async () => {
+    const redis = new FakeRedis();
+    const metrics = new RecordingMetrics();
+    const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, metrics });
+    let version = 1;
+    const getUser = dialcache.cached(async (userId: string) => ({ userId, version }), {
+      keyType: "user_id",
+      useCase: "MissReasonClassification",
+      cacheKey: (userId) => userId,
+      trackForInvalidation: true,
+      defaultConfig: remoteOnly(),
+    });
+    const labels = {
+      cacheNamespace: "urn",
+      useCase: "MissReasonClassification",
+      keyType: "user_id",
+      layer: CacheLayer.REMOTE,
+    };
+    const remoteMisses = () =>
+      metrics.events.filter((event) => event.name === "miss" && event.labels.layer === CacheLayer.REMOTE);
+
+    await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ userId: "123", version: 1 });
+    expect(remoteMisses()).toEqual([{ name: "miss", labels: { ...labels, reason: "not_found" } }]);
+
+    version = 2;
+    await dialcache.invalidateRemote("user_id", "123");
+    vi.advanceTimersByTime(1);
+
+    await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ userId: "123", version: 2 });
+    expect(remoteMisses()).toEqual([
+      { name: "miss", labels: { ...labels, reason: "not_found" } },
+      { name: "miss", labels: { ...labels, reason: "watermark_invalidated" } },
+    ]);
+
+    // The fenced read refilled Redis, so the next read hits without a new miss.
+    await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ userId: "123", version: 2 });
+    expect(remoteMisses()).toHaveLength(2);
   });
 
   it("preserves the furthest watermark across repeated invalidations", async () => {
diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts
index dc20218..1ec15b0 100644
--- a/test/dialcache-liveness.test.ts
+++ b/test/dialcache-liveness.test.ts
@@ -11,6 +11,7 @@ import {
   type CachedOptions,
   type DialCacheMetricsAdapter,
   type DialCacheRedisClient,
+  type RedisReadOutcome,
   type Serializer,
 } from "../src/index.js";
 import { FakeRedis } from "./fake-redis.js";
@@ -538,7 +539,7 @@ describe("DialCache fallback liveness", () => {
   });
 
   it("uses a separate remote-read deadline instead of the fallback deadline", async () => {
-    const readGate = deferred<null>();
+    const readGate = deferred<RedisReadOutcome>();
     const readStarted = deferred<void>();
     const fallback = vi.fn(async () => "value");
     const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
@@ -568,7 +569,7 @@ describe("DialCache fallback liveness", () => {
     expect(vi.getTimerCount()).toBe(1);
     expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1);
 
-    readGate.resolve(null);
+    readGate.resolve({ status: "miss", reason: "not_found" });
     await expect(result).resolves.toBe("value");
     expect(fallback).toHaveBeenCalledTimes(1);
     expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0);
@@ -588,7 +589,7 @@ describe("DialCache fallback liveness", () => {
       },
     };
     const redis: DialCacheRedisClient = {
-      read: async () => "stored",
+      read: async () => ({ status: "hit", payload: "stored" }),
       write: async () => true,
       invalidate: async () => undefined,
     };
@@ -632,7 +633,7 @@ describe("DialCache fallback liveness", () => {
       load: (value) => value.toString(),
     };
     const redis: DialCacheRedisClient = {
-      read: async () => null,
+      read: async () => ({ status: "miss", reason: "not_found" }),
       write: async () => {
         writeStarted.resolve();
         return await writeGate.promise;
diff --git a/test/dialcache-logger.test.ts b/test/dialcache-logger.test.ts
index a7df3e1..ed53e7c 100644
--- a/test/dialcache-logger.test.ts
+++ b/test/dialcache-logger.test.ts
@@ -7,6 +7,7 @@ import {
   DialCacheKeyConfig,
   type DialCacheRedisClient,
   type Logger,
+  type RedisReadOutcome,
 } from "../src/index.js";
 import { FakeRedis } from "./fake-redis.js";
 
@@ -159,7 +160,7 @@ describe("DialCache logger isolation", () => {
     const logger = throwingLogger();
     const invalidationError = new Error("invalidation failed");
     const redis = {
-      read: vi.fn(async () => null),
+      read: vi.fn(async (): Promise<RedisReadOutcome> => ({ status: "miss", reason: "not_found" })),
       write: vi.fn(async () => true),
       invalidate: vi.fn(async () => {
         throw invalidationError;
diff --git a/test/dialcache-metrics.test.ts b/test/dialcache-metrics.test.ts
index 15f5fa1..1ac9e7b 100644
--- a/test/dialcache-metrics.test.ts
+++ b/test/dialcache-metrics.test.ts
@@ -113,7 +113,7 @@ describe("DialCache observability metrics", () => {
     };
 
     isolatedMetrics.request(labels);
-    isolatedMetrics.miss(labels);
+    isolatedMetrics.miss({ ...labels, reason: "not_found" });
     isolatedMetrics.disabled({ ...labels, reason: "ramped_down" });
     isolatedMetrics.error({ ...labels, error: "cache_read", inFallback: false });
     isolatedMetrics.invalidation({
@@ -232,7 +232,9 @@ describe("DialCache observability metrics", () => {
     expect(first).toEqual({ userId: "123", calls: 1 });
     expect(second).toEqual({ userId: "123", calls: 1 });
     expect(events(metrics, "request", { useCase: "CustomMetricsAdapter", layer: CacheLayer.LOCAL })).toHaveLength(2);
-    expect(events(metrics, "miss", { useCase: "CustomMetricsAdapter", layer: CacheLayer.LOCAL })).toHaveLength(1);
+    expect(
+      events(metrics, "miss", { useCase: "CustomMetricsAdapter", layer: CacheLayer.LOCAL, reason: "not_found" }),
+    ).toHaveLength(1);
     expect(events(metrics, "fallback", { useCase: "CustomMetricsAdapter", layer: CacheLayer.LOCAL })).toHaveLength(1);
     expect(events(metrics, "get", { useCase: "CustomMetricsAdapter", layer: CacheLayer.LOCAL })).toHaveLength(2);
   });
@@ -257,7 +259,9 @@ describe("DialCache observability metrics", () => {
     expect(values[2]).toBe(values[0]);
     expect(calls).toBe(1);
     expect(events(metrics, "request", { useCase: "RequestLocalMetrics", layer: "request_local" })).toHaveLength(2);
-    expect(events(metrics, "miss", { useCase: "RequestLocalMetrics", layer: "request_local" })).toHaveLength(1);
+    expect(
+      events(metrics, "miss", { useCase: "RequestLocalMetrics", layer: "request_local", reason: "not_found" }),
+    ).toHaveLength(1);
     expect(events(metrics, "get", { useCase: "RequestLocalMetrics", layer: "request_local" })).toHaveLength(2);
     expect(events(metrics, "fallback", { useCase: "RequestLocalMetrics", layer: "request_local" })).toHaveLength(1);
     expect(events(metrics, "coalesced", { useCase: "RequestLocalMetrics", scope: "request_local" })).toHaveLength(1);
@@ -589,6 +593,7 @@ describe("DialCache observability metrics", () => {
       events(metrics, "miss", {
         useCase: "SerializationLoadClassification",
         layer: CacheLayer.REMOTE,
+        reason: "deserialization_failed",
       }),
     ).toHaveLength(1);
     expect(JSON.stringify(events(metrics, "error", {}))).not.toMatch(
diff --git a/test/dialcache-redis-read-deadline.test.ts b/test/dialcache-redis-read-deadline.test.ts
index acc57bd..f3fc651 100644
--- a/test/dialcache-redis-read-deadline.test.ts
+++ b/test/dialcache-redis-read-deadline.test.ts
@@ -11,11 +11,13 @@ import {
   type CachedOptions,
   type DialCacheMetricsAdapter,
   type DialCacheRedisClient,
-  type RedisCachePayload,
   type RedisConfig,
   type RedisReadContext,
+  type RedisReadOutcome,
 } from "../src/index.js";
 
+const MISS_NOT_FOUND = { status: "miss", reason: "not_found" } as const;
+
 interface Deferred<T> {
   readonly promise: Promise<T>;
   resolve: (value: T) => void;
@@ -97,7 +99,7 @@ describe("DialCache Redis read deadlines", () => {
         throw new Error("missing read context");
       }
       contexts.push(context);
-      return null;
+      return MISS_NOT_FOUND;
     });
     const libraryDefault = new DialCache({ redis: { client: redis.client } });
     const defaulted = libraryDefault.cached(async () => "defaulted", {
@@ -150,7 +152,7 @@ describe("DialCache Redis read deadlines", () => {
   });
 
   it("accepts an omitted instance default and rejects invalid explicit values", () => {
-    const client = redisClient(async () => null).client;
+    const client = redisClient(async () => MISS_NOT_FOUND).client;
     expect(() => new DialCache({ redis: { client } })).not.toThrow();
 
     const invalidValues: readonly unknown[] = [
@@ -181,7 +183,7 @@ describe("DialCache Redis read deadlines", () => {
   });
 
   it("rejects invalid static use-case overrides before reserving the use-case name", () => {
-    const client = redisClient(async () => null).client;
+    const client = redisClient(async () => MISS_NOT_FOUND).client;
     const invalidValues: readonly unknown[] = [
       null,
       0,
@@ -233,7 +235,7 @@ describe("DialCache Redis read deadlines", () => {
   });
 
   it("fails open before Redis when an explicit runtime timeout is invalid", async () => {
-    const redis = redisClient(async () => null);
+    const redis = redisClient(async () => MISS_NOT_FOUND);
     const error = vi.fn<DialCacheMetricsAdapter["error"]>();
     const metrics = metricsWithError(error);
     const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() };
@@ -286,7 +288,7 @@ describe("DialCache Redis read deadlines", () => {
         if (settlement === "throw") {
           throw new Error("late synchronous Redis failure");
         }
-        return null;
+        return MISS_NOT_FOUND;
       });
       const error = vi.fn<DialCacheMetricsAdapter["error"]>();
       const metrics = metricsWithError(error);
@@ -327,8 +329,8 @@ describe("DialCache Redis read deadlines", () => {
     const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
     const redis = redisClient(
       vi.fn()
-        .mockResolvedValueOnce(JSON.stringify({ source: "redis" }))
-        .mockResolvedValueOnce(null),
+        .mockResolvedValueOnce({ status: "hit", payload: JSON.stringify({ source: "redis" }) })
+        .mockResolvedValueOnce(MISS_NOT_FOUND),
     );
     const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } });
     const hit = dialcache.cached(async () => ({ source: "fallback" }), {
@@ -363,7 +365,7 @@ describe("DialCache Redis read deadlines", () => {
       }
       contexts.push(context);
       readStarted.resolve();
-      return await new Promise<null>(() => undefined);
+      return await new Promise<never>(() => undefined);
     });
     const error = vi.fn<DialCacheMetricsAdapter["error"]>();
     const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() };
@@ -464,7 +466,7 @@ describe("DialCache Redis read deadlines", () => {
           throw new Error("missing read context");
         }
         contexts.push(context);
-        return await new Promise<null>(() => undefined);
+        return await new Promise<never>(() => undefined);
       });
       const dialcache = new DialCache({
         redis: { client: redis.client, readTimeoutMs: 100 },
@@ -503,7 +505,7 @@ describe("DialCache Redis read deadlines", () => {
     const readStarted = deferred<void>();
     const redis = redisClient(async () => {
       readStarted.resolve();
-      return await new Promise<null>(() => undefined);
+      return await new Promise<never>(() => undefined);
     });
     const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 10 } });
     const fallback = vi.fn(async () => "fallback");
@@ -532,7 +534,7 @@ describe("DialCache Redis read deadlines", () => {
     const readStarted = deferred<void>();
     const redis = redisClient(async () => {
       readStarted.resolve();
-      return await new Promise<null>(() => undefined);
+      return await new Promise<never>(() => undefined);
     });
     const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 10 } });
     const fallback = vi.fn(async () => "fallback");
@@ -576,7 +578,7 @@ describe("DialCache Redis read deadlines", () => {
     const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
     const redis = redisClient(async () => {
       readStarted.resolve();
-      return await new Promise<null>(() => undefined);
+      return await new Promise<never>(() => undefined);
     });
     const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 10 } });
     const fallback = vi.fn(async () => "fallback");
@@ -605,13 +607,13 @@ describe("DialCache Redis read deadlines", () => {
   it.each(["fulfillment", "rejection"] as const)(
     "consumes late read %s and lets a later invocation recover",
     async (settlement) => {
-      const firstRead = deferred<RedisCachePayload | null>();
+      const firstRead = deferred<RedisReadOutcome>();
       let readCalls = 0;
       const redis = redisClient(async () => {
         readCalls += 1;
         return readCalls === 1
           ? await firstRead.promise
-          : JSON.stringify({ source: "redis" });
+          : { status: "hit", payload: JSON.stringify({ source: "redis" }) };
       });
       const error = vi.fn<DialCacheMetricsAdapter["error"]>();
       const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() };
@@ -634,7 +636,7 @@ describe("DialCache Redis read deadlines", () => {
       await expect(dialcache.enable(async () => await load())).resolves.toEqual({ source: "redis" });
 
       if (settlement === "fulfillment") {
-        firstRead.resolve(JSON.stringify({ source: "late" }));
+        firstRead.resolve({ status: "hit", payload: JSON.stringify({ source: "late" }) });
       } else {
         firstRead.reject(new Error("late Redis failure"));
       }
@@ -661,7 +663,7 @@ describe("DialCache Redis read deadlines", () => {
           ? async () => {
               throw new Error("Redis unavailable");
             }
-          : async () => await new Promise<null>(() => undefined),
+          : async () => await new Promise<never>(() => undefined),
       );
       const dialcache = new DialCache({
         redis: { client: redis.client, readTimeoutMs: 10 },
@@ -697,7 +699,7 @@ describe("DialCache Redis read deadlines", () => {
 
   it("allocates no read timer for disabled calls, ramped-out Redis, or local hits", async () => {
     const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
-    const redis = redisClient(async () => null);
+    const redis = redisClient(async () => MISS_NOT_FOUND);
     const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } });
     const disabled = dialcache.cached(async () => "disabled", {
       keyType: "id",
diff --git a/test/dialcache-redis.test.ts b/test/dialcache-redis.test.ts
index b59432d..a249d1c 100644
--- a/test/dialcache-redis.test.ts
+++ b/test/dialcache-redis.test.ts
@@ -301,12 +301,12 @@ describe("DialCache Redis TTL layer", () => {
     await redis.write({ valueKey, cacheTtlMs: 60_000, value: payload });
 
     const firstRead = await redis.read({ valueKey });
-    if (!Buffer.isBuffer(firstRead)) {
+    if (firstRead.status !== "hit" || !Buffer.isBuffer(firstRead.payload)) {
       throw new Error("Expected a binary Redis payload");
     }
-    firstRead[0] = 0xff;
+    firstRead.payload[0] = 0xff;
 
-    expect(await redis.read({ valueKey })).toEqual(payload);
+    expect(await redis.read({ valueKey })).toEqual({ status: "hit", payload });
   });
 
   it("fails open when Redis serializer dump fails", async () => {
@@ -419,6 +419,156 @@ describe("DialCache Redis TTL layer", () => {
     expect(logger.warn).not.toHaveBeenCalledWith("Error getting value from Redis cache", expect.any(Error));
   });
 
+  it.each([
+    { shape: "null", outcome: null },
+    { shape: "a bare string payload", outcome: "raw-payload" },
+    { shape: "a bare Buffer payload", outcome: Buffer.from([1, 2, 3]) },
+    { shape: "a non-object value", outcome: 42 },
+    { shape: "an unknown status", outcome: { status: "nope" } },
+    { shape: "an unbounded miss reason", outcome: { status: "miss", reason: "because" } },
+    { shape: "a core-authoritative miss reason", outcome: { status: "miss", reason: "deserialization_failed" } },
+    { shape: "a prototype-chain miss reason", outcome: { status: "miss", reason: "toString" } },
+    { shape: "a non-payload hit", outcome: { status: "hit", payload: 42 } },
+  ])("fails open when a client read returns $shape instead of a read outcome", async ({ outcome }) => {
+    const redisClient: DialCacheRedisClient = {
+      read: vi.fn(async () => outcome as never),
+      write: vi.fn(async () => true),
+      invalidate: vi.fn(async () => undefined),
+    };
+    const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() };
+    const metrics = {
+      request: vi.fn(),
+      miss: vi.fn(),
+      disabled: vi.fn(),
+      error: vi.fn(),
+      invalidation: vi.fn(),
+      observeGet: vi.fn(),
+      observeFallback: vi.fn(),
+      observeSerialization: vi.fn(),
+      observeSize: vi.fn(),
+    };
+    const dialcache = new DialCache({ redis: { client: redisClient, readTimeoutMs: 1_000 }, logger, metrics });
+    let calls = 0;
+    const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), {
+      keyType: "user_id",
+      useCase: "RedisMalformedReadOutcome",
+      cacheKey: (userId) => userId,
+      defaultConfig: new DialCacheKeyConfig({
+        ttlSec: { [CacheLayer.REMOTE]: 60 },
+        ramp: { [CacheLayer.REMOTE]: 100 },
+      }),
+    });
+
+    // A malformed outcome must fail loudly into the fail-open read-error path
+    // instead of being mislabeled as a miss with an unbounded reason.
+    await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ userId: "123", calls: 1 });
+    expect(metrics.miss).not.toHaveBeenCalled();
+    expect(metrics.error).toHaveBeenCalledWith({
+      cacheNamespace: "urn",
+      useCase: "RedisMalformedReadOutcome",
+      keyType: "user_id",
+      layer: CacheLayer.REMOTE,
+      error: "cache_read",
+      inFallback: false,
+    });
+    expect(redisClient.write).not.toHaveBeenCalled();
+    expect(logger.warn).toHaveBeenCalledWith("Error getting value from Redis cache", expect.any(Error));
+  });
+
+  it("emits only the validated reason when a client miss changes answers between reads", async () => {
+    // The outcome guard normalizes onto canonical singletons, so an
+    // accessor-backed miss cannot answer one reason at validation and leak an
+    // unbounded label at metric emission.
+    let reasonReads = 0;
+    const flapping = {
+      status: "miss",
+      get reason(): string {
+        reasonReads += 1;
+        return reasonReads === 1 ? "not_found" : "Tenant123Unbounded";
+      },
+    };
+    const redisClient: DialCacheRedisClient = {
+      read: vi.fn(async () => flapping as never),
+      write: vi.fn(async () => true),
+      invalidate: vi.fn(async () => undefined),
+    };
+    const metrics = {
+      request: vi.fn(),
+      miss: vi.fn(),
+      disabled: vi.fn(),
+      error: vi.fn(),
+      invalidation: vi.fn(),
+      observeGet: vi.fn(),
+      observeFallback: vi.fn(),
+      observeSerialization: vi.fn(),
+      observeSize: vi.fn(),
+    };
+    const dialcache = new DialCache({ redis: { client: redisClient, readTimeoutMs: 1_000 }, metrics });
+    const getUser = dialcache.cached(async (userId: string) => ({ userId }), {
+      keyType: "user_id",
+      useCase: "RedisFlappingMissReason",
+      cacheKey: (userId) => userId,
+      defaultConfig: new DialCacheKeyConfig({
+        ttlSec: { [CacheLayer.REMOTE]: 60 },
+        ramp: { [CacheLayer.REMOTE]: 100 },
+      }),
+    });
+
+    await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ userId: "123" });
+    expect(metrics.miss).toHaveBeenCalledTimes(1);
+    expect(metrics.miss).toHaveBeenCalledWith({
+      cacheNamespace: "urn",
+      useCase: "RedisFlappingMissReason",
+      keyType: "user_id",
+      layer: CacheLayer.REMOTE,
+      reason: "not_found",
+    });
+    expect(metrics.error).not.toHaveBeenCalled();
+  });
+
+  it("serves the payload captured at validation when a client hit changes answers between reads", async () => {
+    let payloadReads = 0;
+    const flapping = {
+      status: "hit",
+      get payload(): string {
+        payloadReads += 1;
+        return payloadReads === 1 ? JSON.stringify({ source: "redis" }) : "garbage";
+      },
+    };
+    const redisClient: DialCacheRedisClient = {
+      read: vi.fn(async () => flapping as never),
+      write: vi.fn(async () => true),
+      invalidate: vi.fn(async () => undefined),
+    };
+    const metrics = {
+      request: vi.fn(),
+      miss: vi.fn(),
+      disabled: vi.fn(),
+      error: vi.fn(),
+      invalidation: vi.fn(),
+      observeGet: vi.fn(),
+      observeFallback: vi.fn(),
+      observeSerialization: vi.fn(),
+      observeSize: vi.fn(),
+    };
+    const fallback = vi.fn(async () => ({ source: "fallback" }));
+    const dialcache = new DialCache({ redis: { client: redisClient, readTimeoutMs: 1_000 }, metrics });
+    const getValue = dialcache.cached(fallback, {
+      keyType: "user_id",
+      useCase: "RedisFlappingHitPayload",
+      cacheKey: () => "123",
+      defaultConfig: new DialCacheKeyConfig({
+        ttlSec: { [CacheLayer.REMOTE]: 60 },
+        ramp: { [CacheLayer.REMOTE]: 100 },
+      }),
+    });
+
+    await expect(dialcache.enable(async () => await getValue())).resolves.toEqual({ source: "redis" });
+    expect(fallback).not.toHaveBeenCalled();
+    expect(metrics.miss).not.toHaveBeenCalled();
+    expect(payloadReads).toBe(1);
+  });
+
   it("records a distinct metric label when a Redis adapter reports invalid payload encoding", async () => {
     const redisClient: DialCacheRedisClient = {
       read: vi.fn(async () => {
diff --git a/test/dialcache-shadow-confirmation.test.ts b/test/dialcache-shadow-confirmation.test.ts
index c024aef..8f6cf3c 100644
--- a/test/dialcache-shadow-confirmation.test.ts
+++ b/test/dialcache-shadow-confirmation.test.ts
@@ -19,6 +19,7 @@ import {
   type RedisCachePayload,
   type RedisInvalidationRequest,
   type RedisReadContext,
+  type RedisReadOutcome,
   type RedisReadRequest,
   type RedisWriteRequest,
   type SerializationMetricLabels,
@@ -52,7 +53,20 @@ function deferred<T>(): Deferred<T> {
   return { promise, resolve, reject };
 }
 
-type ReadStep = () => RedisCachePayload | null | Promise<RedisCachePayload | null>;
+type ReadStepResult = RedisCachePayload | null | RedisReadOutcome;
+type ReadStep = () => ReadStepResult | Promise<ReadStepResult>;
+
+// Scripted payloads stay terse: null means a not_found miss, a payload means a
+// hit, and an explicit RedisReadOutcome passes through for reason-specific steps.
+function normalizeReadStep(result: ReadStepResult): RedisReadOutcome {
+  if (result === null) {
+    return { status: "miss", reason: "not_found" };
+  }
+  if (typeof result === "string" || Buffer.isBuffer(result)) {
+    return { status: "hit", payload: result };
+  }
+  return result;
+}
 
 class ScriptedRedis implements DialCacheRedisClient {
   readonly requests: RedisReadRequest[] = [];
@@ -62,14 +76,14 @@ class ScriptedRedis implements DialCacheRedisClient {
 
   constructor(private readonly steps: ReadStep[]) {}
 
-  async read(request: RedisReadRequest, context?: RedisReadContext): Promise<RedisCachePayload | null> {
+  async read(request: RedisReadRequest, context?: RedisReadContext): Promise<RedisReadOutcome> {
     this.requests.push(request);
     this.contexts.push(context);
     const step = this.steps.shift();
     if (step === undefined) {
       throw new Error("Unexpected Redis read");
     }
-    return await step();
+    return normalizeReadStep(await step());
   }
 }
 
@@ -1002,6 +1016,37 @@ describe("DialCache Redis shadow confirmation", () => {
     )).toHaveLength(0);
   });
 
+  it.each([
+    { name: "fills", wroteRemote: true, expected: "filled" },
+    { name: "reports fill_blocked", wroteRemote: false, expected: "fill_blocked" },
+  ])("$name after a watermark-fenced dark read, keeping fenced misses fill-eligible", async ({
+    wroteRemote,
+    expected,
+  }) => {
+    // A fenced dark read must stay fill-eligible: the tracked write script
+    // re-checks the watermark server-side, so eligibility never changes here.
+    const redis = new ScriptedRedis([() => ({ status: "miss", reason: "watermark_invalidated" })]);
+    redis.write.mockImplementationOnce(async () => wroteRemote);
+    const metrics = new RecordingMetrics();
+    const dialcache = createCache(redis, metrics);
+    const getUser = dialcache.cached(async () => ({ id: "123" }), {
+      ...trackedOptions(`ShadowFencedDarkRead${expected}`, remoteConfig(0)),
+      cacheKey: () => "123",
+    });
+
+    await expect(dialcache.enable(async () => await getUser())).resolves.toEqual({ id: "123" });
+    await waitForShadowEvents(metrics, 1);
+
+    expect(metrics.shadowEvents.map(({ outcome }) => outcome)).toEqual([expected]);
+    expect(redis.write).toHaveBeenCalledOnce();
+    expect(redis.write).toHaveBeenCalledWith(expect.objectContaining({ watermarkKey: expect.any(String) }));
+    expect(metrics.ordinaryEvents.filter(({ name, labels }) =>
+      name === "miss"
+      && labels.layer === REMOTE_SHADOW_CACHE_LAYER
+      && labels.reason === "watermark_invalidated"
+    )).toHaveLength(1);
+  });
+
   it("reports a detached serializer dump failure as fill_error with an exact remote_shadow error", async () => {
     const redis = new ScriptedRedis([() => null]);
     const metrics = new RecordingMetrics();
diff --git a/test/fake-redis.ts b/test/fake-redis.ts
index 93f5102..32e3d09 100644
--- a/test/fake-redis.ts
+++ b/test/fake-redis.ts
@@ -2,10 +2,11 @@ import type {
   DialCacheRedisClient,
   RedisCachePayload,
   RedisInvalidationRequest,
+  RedisReadOutcome,
   RedisReadRequest,
   RedisWriteRequest,
 } from "../src/index.js";
-import { DialCacheRedisPayloadEncodingError } from "../src/redis-client.js";
+import { decodeRedisFrame, decodeTrackedRedisFrame } from "../src/redis-protocol.js";
 
 const FRAME_VERSION = 1;
 const ENCODING_OFFSET = 9;
@@ -27,7 +28,7 @@ export class FakeRedis implements DialCacheRedisClient {
   failWatermarkGet = false;
   getGate: Promise<void> | null = null;
 
-  async read({ valueKey, watermarkKey }: RedisReadRequest): Promise<RedisCachePayload | null> {
+  async read({ valueKey, watermarkKey }: RedisReadRequest): Promise<RedisReadOutcome> {
     if (watermarkKey === undefined) {
       this.getCalls += 1;
     } else {
@@ -122,32 +123,15 @@ export class FakeRedis implements DialCacheRedisClient {
     }
   }
 
-  private readPayload(valueKey: string, watermarkKey: string | null): RedisCachePayload | null {
-    const raw = this.readRaw(valueKey);
-    if (raw === null || raw.length < PAYLOAD_OFFSET || raw[0] !== FRAME_VERSION) {
-      return null;
-    }
-
-    if (watermarkKey !== null) {
-      let watermark: number | null;
-      try {
-        watermark = this.readWatermark(watermarkKey);
-      } catch {
-        return null;
-      }
-      if (watermark === null || Number(readTimestamp(raw)) <= watermark) {
-        return null;
-      }
-    }
-
-    const encoding = raw[ENCODING_OFFSET];
-    if (encoding === 0) {
-      return raw.subarray(PAYLOAD_OFFSET).toString("utf8");
-    }
-    if (encoding === 1) {
-      return Buffer.from(raw.subarray(PAYLOAD_OFFSET));
-    }
-    throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding");
+  // Delegates to the shared decoders so classification can never drift from the
+  // bundled adapters. The frame is copied first so the decoders' zero-copy
+  // payload views never alias this fake's persistent store.
+  private readPayload(valueKey: string, watermarkKey: string | null): RedisReadOutcome {
+    const stored = this.readRaw(valueKey);
+    const frame = stored === null ? null : Buffer.from(stored);
+    return watermarkKey === null
+      ? decodeRedisFrame(frame)
+      : decodeTrackedRedisFrame(frame, this.readRaw(watermarkKey));
   }
 
   private storeFrame(key: string, ttlMs: number, payload: RedisCachePayload): void {
diff --git a/test/node-redis.test.ts b/test/node-redis.test.ts
index dc740b9..37cc73f 100644
--- a/test/node-redis.test.ts
+++ b/test/node-redis.test.ts
@@ -109,10 +109,10 @@ describe("node-redis adapter", () => {
     });
     const adapter = createNodeRedisDialCacheClient(client as never);
 
-    await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain");
+    await expect(adapter.read({ valueKey: "plain:value" })).resolves.toEqual({ status: "hit", payload: "plain" });
     await expect(
       adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }),
-    ).resolves.toEqual(Buffer.from([0, 0xff]));
+    ).resolves.toEqual({ status: "hit", payload: Buffer.from([0, 0xff]) });
     await expect(
       adapter.write({ valueKey: "plain:value", cacheTtlMs: 1_000, value: "plain" }),
     ).resolves.toBe(true);
@@ -161,7 +161,7 @@ describe("node-redis adapter", () => {
     await expect(adapter.read(
       { valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" },
       { timeoutMs: 25, signal: controller.signal },
-    )).resolves.toBe("tracked");
+    )).resolves.toEqual({ status: "hit", payload: "tracked" });
 
     expect(client.sendCommand).toHaveBeenCalledWith(
       "tracked:{id}:value",
@@ -183,7 +183,7 @@ describe("node-redis adapter", () => {
     await expect(adapter.read({
       valueKey: "tracked:{id}:value",
       watermarkKey: "tracked:{id}:watermark",
-    })).resolves.toBe("tracked");
+    })).resolves.toEqual({ status: "hit", payload: "tracked" });
 
     expect(client.sendCommand).toHaveBeenCalledWith(
       ["MGET", "tracked:{id}:value", "tracked:{id}:watermark"],
diff --git a/test/prometheus.test.ts b/test/prometheus.test.ts
index 89bbc62..2006177 100644
--- a/test/prometheus.test.ts
+++ b/test/prometheus.test.ts
@@ -15,6 +15,7 @@ import {
   type CompressionOutcome,
   type DisabledReason,
   type MetricErrorKind,
+  type MissReason,
   type ShadowValidationOutcome,
 } from "../src/index.js";
 import { PrometheusDialCacheMetrics, createPrometheusDialCacheMetrics } from "../src/prometheus.js";
@@ -58,6 +59,13 @@ const DISABLED_REASONS: Readonly<Record<DisabledReason, true>> = {
   ramped_down: true,
   config_error: true,
 };
+const MISS_REASONS: Readonly<Record<MissReason, true>> = {
+  not_found: true,
+  frame_unsupported: true,
+  watermark_unreadable: true,
+  watermark_invalidated: true,
+  deserialization_failed: true,
+};
 const COMPRESSION_OUTCOMES: Readonly<Record<CompressionOutcome, true>> = {
   compressed: true,
   below_threshold: true,
@@ -166,7 +174,7 @@ describe("Prometheus metrics adapter", () => {
     } as const;
 
     metrics.request(labels);
-    metrics.miss(labels);
+    metrics.miss({ ...labels, reason: "not_found" });
     metrics.disabled({ ...labels, reason: "context" });
     metrics.error({ ...labels, error: "cache_read", inFallback: false });
     metrics.invalidation({ cacheNamespace: labels.cacheNamespace, keyType: labels.keyType, layer: labels.layer });
@@ -226,7 +234,7 @@ describe("Prometheus metrics adapter", () => {
       histogramSchema("schema_dialcache_fallback_timer", ["cache_namespace", "use_case", "key_type", "layer"], TIMER_BUCKETS),
       histogramSchema("schema_dialcache_get_timer", ["cache_namespace", "use_case", "key_type", "layer"], TIMER_BUCKETS),
       counterSchema("schema_dialcache_invalidation_counter", ["cache_namespace", "key_type", "layer"]),
-      counterSchema("schema_dialcache_miss_counter", ["cache_namespace", "use_case", "key_type", "layer"]),
+      counterSchema("schema_dialcache_miss_counter", ["cache_namespace", "use_case", "key_type", "layer", "reason"]),
       counterSchema("schema_dialcache_request_counter", ["cache_namespace", "use_case", "key_type", "layer"]),
       histogramSchema(
         "schema_dialcache_serialization_timer",
@@ -324,6 +332,34 @@ describe("Prometheus metrics adapter", () => {
     }
   });
 
+  it("exports every bounded miss reason without rewriting labels", async () => {
+    const registry = new Registry();
+    const metrics = new PrometheusDialCacheMetrics({ registry, prefix: "miss_reason_" });
+    const labels = {
+      cacheNamespace: "users",
+      useCase: "PrometheusMissReasons",
+      keyType: "user_id",
+      layer: CacheLayer.REMOTE,
+    } as const;
+
+    const missReasons = Object.keys(MISS_REASONS) as MissReason[];
+    for (const reason of missReasons) {
+      metrics.miss({ ...labels, reason });
+    }
+
+    for (const reason of missReasons) {
+      await expect(
+        sumMetric(registry, "miss_reason_dialcache_miss_counter", {
+          cache_namespace: labels.cacheNamespace,
+          use_case: labels.useCase,
+          key_type: labels.keyType,
+          layer: labels.layer,
+          reason,
+        }),
+      ).resolves.toBe(1);
+    }
+  });
+
   it("exports every bounded shadow-validation outcome without adding cache identity or layer labels", async () => {
     const registry = new Registry();
     const metrics = new PrometheusDialCacheMetrics({ registry, prefix: "shadow_" });
@@ -396,7 +432,7 @@ describe("Prometheus metrics adapter", () => {
     } as const;
 
     metrics.request(labels);
-    metrics.miss(labels);
+    metrics.miss({ ...labels, reason: "watermark_invalidated" });
     metrics.observeGet(labels, 0.01);
 
     await expect(
diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts
index 67fbfdc..43ba2f6 100644
--- a/test/redis-cluster.integration.test.ts
+++ b/test/redis-cluster.integration.test.ts
@@ -214,7 +214,7 @@ describe("DialCache Redis protocol on Redis Cluster", () => {
     const payload = Buffer.from(Array.from({ length: 256 }, (_, index) => index));
 
     expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true);
-    expect(await scriptClient.read({ valueKey })).toEqual(payload);
+    expect(await scriptClient.read({ valueKey })).toEqual({ status: "hit", payload });
 
     const stored = await cluster.get(commandOptions({ returnBuffers: true }), valueKey);
     expect(stored?.length).toBe(10 + payload.length);
@@ -232,6 +232,9 @@ describe("DialCache Redis protocol on Redis Cluster", () => {
         value: trackedPayload,
       }),
     ).toBe(true);
-    expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload);
+    expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual({
+      status: "hit",
+      payload: trackedPayload,
+    });
   });
 });
diff --git a/test/redis-payload.test.ts b/test/redis-payload.test.ts
index d291360..2cdffab 100644
--- a/test/redis-payload.test.ts
+++ b/test/redis-payload.test.ts
@@ -25,24 +25,41 @@ function encodeFrame(
 
 describe("Redis frame decoding", () => {
   it("decodes UTF-8 and binary payloads without copying binary data", () => {
-    expect(decodeRedisFrame(encodeFrame("cached"))).toBe("cached");
+    expect(decodeRedisFrame(encodeFrame("cached"))).toEqual({ status: "hit", payload: "cached" });
 
     const frame = encodeFrame(Buffer.from([0, 0xff, 0x80]), 1);
-    const decoded = decodeRedisFrame(frame);
-    expect(decoded).toEqual(Buffer.from([0, 0xff, 0x80]));
-    expect(Buffer.isBuffer(decoded)).toBe(true);
-    if (!Buffer.isBuffer(decoded)) {
+    const outcome = decodeRedisFrame(frame);
+    expect(outcome).toEqual({ status: "hit", payload: Buffer.from([0, 0xff, 0x80]) });
+    if (outcome.status !== "hit" || !Buffer.isBuffer(outcome.payload)) {
       throw new Error("Expected a binary Redis payload");
     }
+    const decoded = outcome.payload;
     expect(decoded.buffer).toBe(frame.buffer);
     expect(decoded.byteOffset).toBe(frame.byteOffset + 10);
     expect(decoded.byteLength).toBe(frame.byteLength - 10);
   });
 
-  it("treats missing, short, and unsupported frames as misses", () => {
-    expect(decodeRedisFrame(null)).toBeNull();
-    expect(decodeRedisFrame(Buffer.alloc(9))).toBeNull();
-    expect(decodeRedisFrame(encodeFrame("cached", 0, 1_000, 2))).toBeNull();
+  it("shares one frozen canonical outcome per miss reason that consumers cannot corrupt", () => {
+    const outcome = decodeRedisFrame(null);
+
+    // The decoders and the core's normalizer alias these singletons, so their
+    // immutability is what keeps one consumer's mutation from corrupting
+    // every later miss (and its bounded metric label) process-wide.
+    expect(decodeRedisFrame(null)).toBe(outcome);
+    expect(Object.isFrozen(outcome)).toBe(true);
+    expect(() => {
+      (outcome as { reason: string }).reason = "corrupted";
+    }).toThrow(TypeError);
+    expect(decodeRedisFrame(null)).toEqual({ status: "miss", reason: "not_found" });
+  });
+
+  it("classifies missing frames apart from short and unsupported frames", () => {
+    expect(decodeRedisFrame(null)).toEqual({ status: "miss", reason: "not_found" });
+    expect(decodeRedisFrame(Buffer.alloc(9))).toEqual({ status: "miss", reason: "frame_unsupported" });
+    expect(decodeRedisFrame(encodeFrame("cached", 0, 1_000, 2))).toEqual({
+      status: "miss",
+      reason: "frame_unsupported",
+    });
   });
 
   it("rejects unsupported payload encodings after validating the frame", () => {
@@ -64,13 +81,19 @@ describe("Redis frame decoding", () => {
   it("validates tracked frames against integer and fractional watermarks", () => {
     const frame = encodeFrame("cached", 0, 1_000);
 
-    expect(decodeTrackedRedisFrame(frame, Buffer.from("999"))).toBe("cached");
-    expect(decodeTrackedRedisFrame(frame, Buffer.from("999.5"))).toBe("cached");
-    expect(decodeTrackedRedisFrame(frame, Buffer.from("1000"))).toBeNull();
-    expect(decodeTrackedRedisFrame(frame, Buffer.from("1000.5"))).toBeNull();
+    expect(decodeTrackedRedisFrame(frame, Buffer.from("999"))).toEqual({ status: "hit", payload: "cached" });
+    expect(decodeTrackedRedisFrame(frame, Buffer.from("999.5"))).toEqual({ status: "hit", payload: "cached" });
+    expect(decodeTrackedRedisFrame(frame, Buffer.from("1000"))).toEqual({
+      status: "miss",
+      reason: "watermark_invalidated",
+    });
+    expect(decodeTrackedRedisFrame(frame, Buffer.from("1000.5"))).toEqual({
+      status: "miss",
+      reason: "watermark_invalidated",
+    });
   });
 
-  it("treats missing, malformed, and non-finite watermarks as misses", () => {
+  it("treats missing, malformed, and non-finite watermarks as unreadable misses", () => {
     const frame = encodeFrame("cached", 0, 1_000);
 
     for (const watermark of [
@@ -83,17 +106,38 @@ describe("Redis frame decoding", () => {
       Buffer.from("1\n"),
       Buffer.from("9".repeat(400)),
     ]) {
-      expect(decodeTrackedRedisFrame(frame, watermark)).toBeNull();
+      expect(decodeTrackedRedisFrame(frame, watermark)).toEqual({
+        status: "miss",
+        reason: "watermark_unreadable",
+      });
     }
   });
 
   it("validates tracked frame and watermark state before payload encoding", () => {
     const malformedPayload = encodeFrame("cached", 2, 1_000);
 
-    expect(decodeTrackedRedisFrame(null, Buffer.from("0"))).toBeNull();
-    expect(decodeTrackedRedisFrame(Buffer.alloc(9), Buffer.from("0"))).toBeNull();
-    expect(decodeTrackedRedisFrame(malformedPayload, null)).toBeNull();
-    expect(decodeTrackedRedisFrame(malformedPayload, Buffer.from("1000"))).toBeNull();
+    // Frame state wins over watermark state even when both are bad.
+    expect(decodeTrackedRedisFrame(null, Buffer.from("not-a-watermark"))).toEqual({
+      status: "miss",
+      reason: "not_found",
+    });
+    expect(decodeTrackedRedisFrame(Buffer.alloc(9), Buffer.from("not-a-watermark"))).toEqual({
+      status: "miss",
+      reason: "frame_unsupported",
+    });
+    expect(decodeTrackedRedisFrame(null, Buffer.from("0"))).toEqual({ status: "miss", reason: "not_found" });
+    expect(decodeTrackedRedisFrame(Buffer.alloc(9), Buffer.from("0"))).toEqual({
+      status: "miss",
+      reason: "frame_unsupported",
+    });
+    expect(decodeTrackedRedisFrame(malformedPayload, null)).toEqual({
+      status: "miss",
+      reason: "watermark_unreadable",
+    });
+    expect(decodeTrackedRedisFrame(malformedPayload, Buffer.from("1000"))).toEqual({
+      status: "miss",
+      reason: "watermark_invalidated",
+    });
     expect(() => decodeTrackedRedisFrame(malformedPayload, Buffer.from("999"))).toThrow(
       DialCacheRedisPayloadEncodingError,
     );
diff --git a/test/redis-real.integration.test.ts b/test/redis-real.integration.test.ts
index d5508ae..73aaf2d 100644
--- a/test/redis-real.integration.test.ts
+++ b/test/redis-real.integration.test.ts
@@ -354,8 +354,8 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
         const roundTrip = await scriptClient.read({ valueKey });
         const stored = await admin.get(commandOptions({ returnBuffers: true }), valueKey);
 
-        expect(Buffer.isBuffer(roundTrip)).toBe(true);
-        expect(roundTrip).toEqual(payload);
+        expect(roundTrip).toEqual({ status: "hit", payload });
+        expect(roundTrip.status === "hit" && Buffer.isBuffer(roundTrip.payload)).toBe(true);
         expect(stored).not.toBeNull();
         expect(stored?.length).toBe(10 + payload.length);
         expect(stored?.[0]).toBe(1);
@@ -374,7 +374,10 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
           value: trackedPayload,
         }),
       ).toBe(true);
-      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload);
+      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual({
+        status: "hit",
+        payload: trackedPayload,
+      });
     });
 
     it("shadow-validates the deserialized tracked value without repairing a mismatch", async () => {
@@ -759,7 +762,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       expect(await client.adapter.read({
         valueKey,
         ...(tracked ? { watermarkKey } : {}),
-      })).toBe(JSON.stringify(sourceValue));
+      })).toEqual({ status: "hit", payload: JSON.stringify(sourceValue) });
       expect(await admin.pTTL(valueKey)).toBeGreaterThan(55_000);
       expect(await admin.pTTL(valueKey)).toBeLessThanOrEqual(60_000);
       if (tracked) {
@@ -787,6 +790,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
         useCase,
         keyType: "item_id",
         layer: "remote_shadow",
+        reason: "not_found",
       });
       expect(metrics.observeSerialization).toHaveBeenCalledWith({
         cacheNamespace: namespace,
@@ -891,7 +895,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
 
       await admin.scriptFlush();
       expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: "untracked" })).toBe(true);
-      expect(await scriptClient.read({ valueKey })).toBe("untracked");
+      expect(await scriptClient.read({ valueKey })).toEqual({ status: "hit", payload: "untracked" });
 
       const trackedValueKey = "script-recovery:{item:tracked}:value";
       const watermarkKey = "script-recovery:{item:tracked}:watermark";
@@ -904,7 +908,10 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
           value: "tracked",
         }),
       ).toBe(true);
-      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBe("tracked");
+      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual({
+        status: "hit",
+        payload: "tracked",
+      });
       await admin.scriptFlush();
       await expect(
         scriptClient.invalidate({
@@ -912,10 +919,13 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
           futureBufferMs: 0,
         }),
       ).resolves.toBeUndefined();
-      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_invalidated",
+      });
     });
 
-    it("treats every invalid read frame and watermark state as a miss", async () => {
+    it("classifies every invalid read frame and watermark state with an exact miss reason", async () => {
       if (client === undefined || admin === undefined) {
         throw new Error("Redis test clients did not start");
       }
@@ -923,28 +933,46 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       const valueKey = "read-paths:{item:read}:value";
       const watermarkKey = "read-paths:{item:read}:watermark";
 
-      expect(await scriptClient.read({ valueKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey })).toEqual({ status: "miss", reason: "not_found" });
+
+      // Frame-before-watermark precedence: an absent frame stays not_found
+      // even when the watermark key holds garbage.
+      await admin.set(watermarkKey, "not-a-watermark");
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ status: "miss", reason: "not_found" });
+      await admin.del(watermarkKey);
 
       await admin.set(valueKey, Buffer.alloc(9));
-      expect(await scriptClient.read({ valueKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey })).toEqual({ status: "miss", reason: "frame_unsupported" });
 
       await admin.set(valueKey, encodeFrame("wrong-version", 0, 1_000, 2));
-      expect(await scriptClient.read({ valueKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey })).toEqual({ status: "miss", reason: "frame_unsupported" });
 
       await admin.set(valueKey, encodeFrame("tracked", 0, 1_000));
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_unreadable",
+      });
 
       await admin.set(watermarkKey, "not-a-watermark");
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_unreadable",
+      });
 
       await admin.set(watermarkKey, "9".repeat(400));
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_unreadable",
+      });
 
       await admin.set(watermarkKey, "1000");
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_invalidated",
+      });
 
       await admin.set(watermarkKey, "999.5");
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("tracked");
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ status: "hit", payload: "tracked" });
     });
 
     it("records a stale tracked frame as a remote miss without a read error", async () => {
@@ -999,7 +1027,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       expect(metrics.request).toHaveBeenCalledOnce();
       expect(metrics.request).toHaveBeenCalledWith(labels);
       expect(metrics.miss).toHaveBeenCalledOnce();
-      expect(metrics.miss).toHaveBeenCalledWith(labels);
+      expect(metrics.miss).toHaveBeenCalledWith({ ...labels, reason: "watermark_invalidated" });
       expect(metrics.observeGet).toHaveBeenCalledOnce();
       expect(metrics.observeGet).toHaveBeenCalledWith(labels, expect.any(Number));
       expect(metrics.observeFallback).toHaveBeenCalledOnce();
@@ -1018,12 +1046,18 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       await admin.hSet(valueKey, "field", "value");
       await admin.set(watermarkKey, "0");
       await expect(scriptClient.read({ valueKey })).rejects.toThrow(/WRONGTYPE/);
-      await expect(scriptClient.read({ valueKey, watermarkKey })).resolves.toBeNull();
+      await expect(scriptClient.read({ valueKey, watermarkKey })).resolves.toEqual({
+        status: "miss",
+        reason: "not_found",
+      });
 
       await admin.del([valueKey, watermarkKey]);
       await admin.set(valueKey, encodeFrame("cached", 0, 1_000));
       await admin.hSet(watermarkKey, "field", "value");
-      await expect(scriptClient.read({ valueKey, watermarkKey })).resolves.toBeNull();
+      await expect(scriptClient.read({ valueKey, watermarkKey })).resolves.toEqual({
+        status: "miss",
+        reason: "watermark_unreadable",
+      });
 
       const namespace = "wrong-type-repair";
       const repairValueKey = `{${namespace}:item_id:repair}#WrongTypeRepair:dialcache-frame-v1`;
@@ -1321,7 +1355,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
         await expect(client.raw.writeTracked(valueKey, watermarkKey, 60_000, 0, "replacement")).rejects.toThrow(
           "invalid DialCache watermark",
         );
-        expect(await scriptClient.read({ valueKey })).toBe("original");
+        expect(await scriptClient.read({ valueKey })).toEqual({ status: "hit", payload: "original" });
       }
     });
 
@@ -1472,7 +1506,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       expect(wrote).toBe(true);
       expect(await admin.get(watermarkKey)).toBe("1.75");
       expect(await admin.pTTL(watermarkKey)).toBeGreaterThanOrEqual(61_000);
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("cached");
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ status: "hit", payload: "cached" });
     });
 
     it("does not rewrite sufficient or persistent watermarks on tracked writes", async () => {
@@ -1534,11 +1568,14 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
       expect(ttlAfterWrite).toBeGreaterThanOrEqual(61_000);
 
       await scriptClient.invalidate({ watermarkKey, futureBufferMs: 100 });
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({
+        status: "miss",
+        reason: "watermark_invalidated",
+      });
       const watermarkBeforeBlockedWrite = await admin.get(watermarkKey);
       const watermarkTtlBeforeBlockedWrite = await admin.pTTL(watermarkKey);
       expect(await scriptClient.write({ ...writeRequest, value: "blocked" })).toBe(false);
-      expect(await scriptClient.read({ valueKey })).toBeNull();
+      expect(await scriptClient.read({ valueKey })).toEqual({ status: "miss", reason: "not_found" });
       expect(await admin.get(watermarkKey)).toBe(watermarkBeforeBlockedWrite);
       const watermarkTtlAfterBlockedWrite = await admin.pTTL(watermarkKey);
       expect(watermarkTtlAfterBlockedWrite).toBeGreaterThan(watermarkTtlBeforeBlockedWrite - 1_000);
@@ -1549,7 +1586,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
 
       await new Promise((resolve) => setTimeout(resolve, 110));
       expect(await scriptClient.write({ ...writeRequest, value: "fresh" })).toBe(true);
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("fresh");
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ status: "hit", payload: "fresh" });
     });
 
     it("documents that losing a watermark removes its publication fence", async () => {
@@ -1573,7 +1610,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
 
       expect(await scriptClient.write(staleWrite)).toBe(true);
       expect(await admin.get(watermarkKey)).toBe("0");
-      expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("stale");
+      expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ status: "hit", payload: "stale" });
     });
   });
 
@@ -1587,10 +1624,16 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
     const binary = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]);
 
     await nodeRedis.write({ valueKey: "interop:node-to-glide", cacheTtlMs: 60_000, value: binary });
-    await expect(valkeyGlide.read({ valueKey: "interop:node-to-glide" })).resolves.toEqual(binary);
+    await expect(valkeyGlide.read({ valueKey: "interop:node-to-glide" })).resolves.toEqual({
+      status: "hit",
+      payload: binary,
+    });
 
     await valkeyGlide.write({ valueKey: "interop:glide-to-node", cacheTtlMs: 60_000, value: "hello" });
-    await expect(nodeRedis.read({ valueKey: "interop:glide-to-node" })).resolves.toBe("hello");
+    await expect(nodeRedis.read({ valueKey: "interop:glide-to-node" })).resolves.toEqual({
+      status: "hit",
+      payload: "hello",
+    });
 
     const nodeTrackedValueKey = "interop:{node-tracked}:value";
     const nodeTrackedWatermarkKey = "interop:{node-tracked}:watermark";
@@ -1605,7 +1648,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
         valueKey: nodeTrackedValueKey,
         watermarkKey: nodeTrackedWatermarkKey,
       }),
-    ).resolves.toEqual(binary);
+    ).resolves.toEqual({ status: "hit", payload: binary });
 
     const glideTrackedValueKey = "interop:{glide-tracked}:value";
     const glideTrackedWatermarkKey = "interop:{glide-tracked}:watermark";
@@ -1620,6 +1663,6 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => {
         valueKey: glideTrackedValueKey,
         watermarkKey: glideTrackedWatermarkKey,
       }),
-    ).resolves.toBe("tracked");
+    ).resolves.toEqual({ status: "hit", payload: "tracked" });
   });
 });
diff --git a/test/valkey-glide.test.ts b/test/valkey-glide.test.ts
index ea7bb6f..f421284 100644
--- a/test/valkey-glide.test.ts
+++ b/test/valkey-glide.test.ts
@@ -147,11 +147,14 @@ describe("Valkey GLIDE adapter", () => {
     );
     const adapter = createValkeyGlideDialCacheClient(client, mockGlide);
 
-    await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain");
+    await expect(adapter.read({ valueKey: "plain:value" })).resolves.toEqual({ status: "hit", payload: "plain" });
     await expect(
       adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }),
-    ).resolves.toEqual(Buffer.from([0, 0xff]));
-    await expect(adapter.read({ valueKey: "missing:value" })).resolves.toBeNull();
+    ).resolves.toEqual({ status: "hit", payload: Buffer.from([0, 0xff]) });
+    await expect(adapter.read({ valueKey: "missing:value" })).resolves.toEqual({
+      status: "miss",
+      reason: "not_found",
+    });
 
     expect(client.get).toHaveBeenNthCalledWith(
       1,
@@ -190,7 +193,7 @@ describe("Valkey GLIDE adapter", () => {
         valueKey: "cluster:{id}:value",
         watermarkKey: "cluster:{id}:watermark",
       }),
-    ).resolves.toBe("tracked-cluster");
+    ).resolves.toEqual({ status: "hit", payload: "tracked-cluster" });
 
     expect(client.customCommand).toHaveBeenCalledWith(
       ["MGET", "cluster:{id}:value", "cluster:{id}:watermark"],
@@ -427,7 +430,7 @@ describe("Valkey GLIDE adapter", () => {
     expect(scriptInstances.every((script) => script.release.mock.calls.length === 0)).toBe(true);
 
     resolveRead?.(redisFrame("done"));
-    await expect(read).resolves.toBe("done");
+    await expect(read).resolves.toEqual({ status: "hit", payload: "done" });
     adapter.dispose();
     expect(scriptInstances.every((script) => script.release.mock.calls.length === 1)).toBe(true);
   });