fix: harden the legacy HTTP rollback path

N-06: export one idempotency-key authority from mutation-intent.ts and use it
in the V2 client. A caller-supplied key is validated before credentials, timers
and fetch, and an invalid value is rejected as VALIDATION_REJECTED /
IDEMPOTENCY_KEY_INVALID rather than trimmed, regenerated or dropped, so a keyed
command can no longer replay while sending no key.

N-07: bound the legacy credential wait by the existing attempt controller,
which already carries the total deadline and the caller signal, so a
non-cooperative owner cannot hold the request open and no extra timer is
introduced. The owner receives the operation context, and the failure follows
ownership: deadline to REQUEST_TIMEOUT, caller to REQUEST_ABORTED, and only a
genuine rejection to AUTH_INTEGRATION_FAILURE. None of these paths fetch.

N-08: readBoundedJson delegates to the common bounded reader, so cancel and
releaseLock throws stay isolated inside the closed result, and the V2
content-type mismatch now cancels the response body.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/adapters/http/bounded-json.ts

@@ -1,50 +1,39 @@
+import { decodeJsonBytes, readBoundedBytes } from "./bounded-body-reader.ts";
+
 export type BoundedJsonResult =
   | Readonly<{ ok: true; value: unknown }>
   | Readonly<{ ok: false; code: "RESPONSE_BODY_LIMIT" | "MALFORMED_JSON" }>;
 
+/**
+ * N-08. The V2 compatibility reader delegates to the common bounded reader
+ * instead of maintaining a second stream-reading strategy.
+ *
+ * `bounded-body-reader` already isolates `cancel()` and `releaseLock()` throws
+ * so a cleanup failure cannot escape the closed result. Only the legacy failure
+ * codes are preserved here:
+ *
+ * - `RESPONSE_TOO_LARGE` → `RESPONSE_BODY_LIMIT`
+ * - `RESPONSE_STREAM_FAILURE` / `UTF8_INVALID` / `JSON_INVALID` →
+ *   `MALFORMED_JSON`
+ */
 export async function readBoundedJson(
   response: Response,
   maxBytes: number,
 ): Promise<BoundedJsonResult> {
-  const declaredLength = Number(response.headers.get("content-length"));
-  if (Number.isFinite(declaredLength) && declaredLength > maxBytes) {
-    await response.body?.cancel();
-    return { ok: false, code: "RESPONSE_BODY_LIMIT" };
-  }
-  if (!response.body) return { ok: false, code: "MALFORMED_JSON" };
-
-  const reader = response.body.getReader();
-  const chunks: Uint8Array[] = [];
-  let total = 0;
-  try {
-    while (true) {
-      const next = await reader.read();
-      if (next.done) break;
-      total += next.value.byteLength;
-      if (total > maxBytes) {
-        await reader.cancel();
-        return { ok: false, code: "RESPONSE_BODY_LIMIT" };
-      }
-      chunks.push(next.value);
-    }
-  } catch {
-    return { ok: false, code: "MALFORMED_JSON" };
-  } finally {
-    reader.releaseLock();
-  }
-
-  const bytes = new Uint8Array(total);
-  let offset = 0;
-  for (const chunk of chunks) {
-    bytes.set(chunk, offset);
-    offset += chunk.byteLength;
-  }
-  try {
-    return {
-      ok: true,
-      value: JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes)),
-    };
-  } catch {
-    return { ok: false, code: "MALFORMED_JSON" };
+  const bytes = await readBoundedBytes(response, maxBytes);
+  if (!bytes.ok) {
+    return Object.freeze({
+      ok: false as const,
+      code:
+        bytes.code === "RESPONSE_TOO_LARGE"
+          ? ("RESPONSE_BODY_LIMIT" as const)
+          : ("MALFORMED_JSON" as const),
+    });
   }
+  // An absent body decodes to zero bytes, which is not valid JSON. The legacy
+  // contract reported that as MALFORMED_JSON, and that is preserved.
+  const decoded = decodeJsonBytes(bytes.bytes);
+  return decoded.ok
+    ? Object.freeze({ ok: true as const, value: decoded.value })
+    : Object.freeze({ ok: false as const, code: "MALFORMED_JSON" as const });
 }

src/adapters/http/client.ts

@@ -19,6 +19,10 @@ import {
   statusGroup,
 } from "../../contracts/diagnostics.ts";
 import type { AuthSessionPort } from "../../application/ports/auth-session-port.ts";
+import { isValidIdempotencyKey } from "../../contracts/mutation-intent.ts";
+
+/** Sentinel for a credential wait ended by the attempt lifetime. */
+const ATTEMPT_ABORTED = Symbol("ATTEMPT_ABORTED");
 import type { ClockPort } from "../../application/ports/clock-port.ts";
 import type { DiagnosticsPort } from "../../application/ports/diagnostics-port.ts";
 import type { TelemetryPort } from "../../application/ports/telemetry-port.ts";
@@ -238,22 +242,52 @@ export function createHttpClient(
       }
       return outcome;
     }
+    // N-06. A caller-supplied key is validated before credentials, timers or
+    // fetch. An invalid value is rejected outright rather than trimmed or
+    // replaced, so a keyed command can never replay with no key at all.
     let logicalIdempotencyKey: string | undefined;
-    try {
-      logicalIdempotencyKey =
-        operation.idempotency === "keyed"
-          ? input.idempotencyKey ?? idempotencyKeyFactory()
-          : undefined;
-    } catch {
-      return finalize(
-        {
-          ok: false,
-          error: failure("UNKNOWN_CLIENT_FAILURE", input.operationId, 0, {
-            code: "IDEMPOTENCY_KEY_CREATION_FAILED",
-          }),
-        },
-        "failed",
-      );
+    if (operation.idempotency === "keyed") {
+      if (input.idempotencyKey !== undefined) {
+        if (!isValidIdempotencyKey(input.idempotencyKey)) {
+          return finalize(
+            {
+              ok: false,
+              error: failure("VALIDATION_REJECTED", input.operationId, 0, {
+                code: "IDEMPOTENCY_KEY_INVALID",
+              }),
+            },
+            "failed",
+          );
+        }
+        logicalIdempotencyKey = input.idempotencyKey;
+      } else {
+        let generated: string;
+        try {
+          generated = idempotencyKeyFactory();
+        } catch {
+          return finalize(
+            {
+              ok: false,
+              error: failure("UNKNOWN_CLIENT_FAILURE", input.operationId, 0, {
+                code: "IDEMPOTENCY_KEY_CREATION_FAILED",
+              }),
+            },
+            "failed",
+          );
+        }
+        if (!isValidIdempotencyKey(generated)) {
+          return finalize(
+            {
+              ok: false,
+              error: failure("UNKNOWN_CLIENT_FAILURE", input.operationId, 0, {
+                code: "IDEMPOTENCY_KEY_CREATION_FAILED",
+              }),
+            },
+            "failed",
+          );
+        }
+        logicalIdempotencyKey = generated;
+      }
     }
     let retryCount = 0;
     let recoveryUsed = false;
@@ -568,11 +602,45 @@ export function createHttpClient(
           };
         }
         try {
-          const patch = await authSession.credentialPatch({
-            origin: target.url.origin,
-            method: operation.method,
-            operationId: operation.operationId,
-          });
+          // N-07. The credential owner is bounded by the attempt lifetime that
+          // already carries the total deadline and the caller signal, so a
+          // non-cooperative owner cannot hold the request open and no extra
+          // timer is introduced. A late completion is observed and discarded.
+          const raced = await raceAttemptSignal(
+            Promise.resolve(
+              authSession.credentialPatch(
+                {
+                  origin: target.url.origin,
+                  method: operation.method,
+                  operationId: operation.operationId,
+                },
+                Object.freeze({
+                  signal: controller.signal,
+                  deadlineAtMonotonicMs: deadlineAt,
+                }),
+              ),
+            ),
+            controller.signal,
+          );
+          if (raced === ATTEMPT_ABORTED) {
+            return {
+              ok: false,
+              error: timedOut
+                ? failure(
+                    "REQUEST_TIMEOUT",
+                    operation.operationId,
+                    attempt,
+                    { code: "OPERATION_DEADLINE_EXCEEDED" },
+                  )
+                : failure(
+                    "REQUEST_ABORTED",
+                    operation.operationId,
+                    attempt,
+                    { code: "REQUEST_ABORTED" },
+                  ),
+            };
+          }
+          const patch = raced;
           for (const [name, value] of Object.entries(patch.headers)) {
             const normalized = name.toLowerCase();
             const allowedHeaders =
@@ -584,6 +652,7 @@ export function createHttpClient(
             headers.set(normalized, value);
           }
         } catch {
+          // An ordinary owner rejection stays an integration failure.
           return {
             ok: false,
             error: failure("AUTH_INTEGRATION_FAILURE", operation.operationId, attempt, {
@@ -663,6 +732,28 @@ export function createHttpClient(
 
   return Object.freeze({ execute });
 
+  function raceAttemptSignal<Value>(
+    operation: Promise<Value>,
+    signal: AbortSignal,
+  ): Promise<Value | typeof ATTEMPT_ABORTED> {
+    operation.catch(() => {});
+    if (signal.aborted) return Promise.resolve(ATTEMPT_ABORTED);
+    return new Promise<Value | typeof ATTEMPT_ABORTED>((resolve, reject) => {
+      const onAbort = () => resolve(ATTEMPT_ABORTED);
+      signal.addEventListener("abort", onAbort, { once: true });
+      operation.then(
+        (value) => {
+          signal.removeEventListener("abort", onAbort);
+          resolve(value);
+        },
+        (error: unknown) => {
+          signal.removeEventListener("abort", onAbort);
+          reject(error instanceof Error ? error : new Error("rejected"));
+        },
+      );
+    });
+  }
+
   function withinLogicalDeadline<Value>(
     promise: Promise<Value>,
     deadlineAt: number,
@@ -714,6 +805,10 @@ async function parseResponse(
   const contentType = mediaType(response.headers.get("content-type"));
   const acceptedMedia = operation.responseMediaTypes ?? ["application/json"];
   if (!contentType || !acceptedMedia.includes(contentType)) {
+    // N-08. A rejected response still owns an open body stream. Cancellation is
+    // best-effort cleanup, so it is started but not awaited: the closed result
+    // must not depend on stream teardown settling.
+    void response.body?.cancel().catch(() => {});
     return {
       ok: false,
       error: failure("CONTENT_TYPE_MISMATCH", operation.operationId, attempt, {

src/contracts/mutation-intent.ts

@@ -23,6 +23,39 @@ function validBoundedString(value: unknown, maxBytes: number): value is string {
   );
 }
 
+/**
+ * N-06. The single idempotency-key authority shared by the V2 compatibility
+ * client and the V3 executor.
+ *
+ * A caller-supplied value is never trimmed, regenerated or silently dropped:
+ * an invalid key is a contract violation, because replaying a keyed command
+ * without its key is exactly the unsafe behaviour the key exists to prevent.
+ */
+export function isValidIdempotencyKey(value: unknown): value is string {
+  if (
+    !validBoundedString(
+      value,
+      MUTATION_INTENT_BOUNDS.idempotencyKeyMaxBytes,
+    )
+  ) {
+    return false;
+  }
+  for (const character of value) {
+    const codePoint = character.codePointAt(0) ?? 0;
+    if (codePoint <= 0x1f || (codePoint >= 0x7f && codePoint <= 0x9f)) {
+      return false;
+    }
+  }
+  return true;
+}
+
+export function defineIdempotencyKey(value: unknown): string {
+  if (!isValidIdempotencyKey(value)) {
+    throw new TypeError("Idempotency key is invalid.");
+  }
+  return value;
+}
+
 export function defineMutationIntent(intent: MutationIntent): MutationIntent {
   if (
     !validBoundedString(