fix(request): correct SSE, retry, multipart, and HEAD handling

.agents/friction-log/20260909170520-cli-timeout-test/friction.md

@@ -0,0 +1,24 @@
+---
+title: "CLI timeout test includes TypeScript process startup"
+severity: "minor"
+---
+
+## Expected Behavior
+
+The CLI deadline test measures request time independently of process startup.
+
+## Current Behavior
+
+CI startup took roughly 3.3 seconds and exceeded the 2.5-second assertion in test/request-io.test.ts.
+
+## Possible Solution
+
+Start timing when the local server receives the request; retain the exit-code and single-request assertions.
+
+## Minimal Reproducible Example
+
+Run pnpm test on a CI runner with slow Node/tsx startup.
+
+## Context
+
+Observed in PR #137, CI run 34253485246.

.changelog/request-io.md

@@ -0,0 +1,5 @@
+---
+wallet-cli: patch
+---
+
+Stream SSE JSON incrementally, respect retry deadlines, preserve repeated multipart fields, and route HEAD output consistently.

src/commands/request.ts

@@ -2,6 +2,7 @@ import { createWriteStream } from "node:fs";
 import { File } from "node:buffer";
 import { mkdir, readFile, writeFile } from "node:fs/promises";
 import { basename, dirname } from "node:path";
+import { Writable } from "node:stream";
 import { pipeline } from "node:stream/promises";
 import { setTimeout as sleep } from "node:timers/promises";
 
@@ -451,14 +452,20 @@ async function fetchWithRetries(
     try {
       const response = await fetchWithRedirects(request, options, fetchImpl);
       if (attempt + 1 < attempts && retryStatuses.has(response.status)) {
-        await waitBeforeRetry(response, options, attempt);
+        await response.body?.cancel();
+        await waitBeforeRetry(response, options, attempt, request.init.signal);
         continue;
       }
       return response;
     } catch (error) {
       lastError = error;
-      if (attempt + 1 >= attempts) break;
-      await waitBeforeRetry(undefined, options, attempt);
+      if (request.init.signal?.aborted || attempt + 1 >= attempts) break;
+      try {
+        await waitBeforeRetry(undefined, options, attempt, request.init.signal);
+      } catch (error) {
+        lastError = error;
+        break;
+      }
     }
   }
 
@@ -1445,15 +1452,29 @@ async function writeResponseBody(
   const headerText = includeHeaders ? responseHeaderText(response) : "";
 
   if (options.head) {
-    write(stdout, headerText);
-    if (outputPath) await writeFile(outputPath, headerText);
+    await writeOutput(outputPath, headerText, stdout);
     return;
   }
 
   if (options.sseJson) {
-    const text = await response.text();
-    const body = sseToNdjson(text);
-    await writeOutput(outputPath, `${headerText}${body}`, stdout);
+    if (outputPath) {
+      try {
+        await mkdir(dirname(outputPath), { recursive: true });
+      } catch (error) {
+        await response.body?.cancel().catch(() => undefined);
+        throw error;
+      }
+      await pipeline(
+        (options) => sseToNdjson(response, headerText, options?.signal),
+        createWriteStream(outputPath),
+      );
+    } else if (stdout instanceof Writable) {
+      await pipeline((options) => sseToNdjson(response, headerText, options?.signal), stdout, {
+        end: false,
+      });
+    } else {
+      for await (const chunk of sseToNdjson(response, headerText)) write(stdout, chunk);
+    }
     return;
   }
 
@@ -1720,9 +1741,9 @@ async function appendFormField(form: UndiciFormData, field: string) {
     const file = new File([await readFile(rawPath)], basename(rawPath), {
       type: contentType ?? "application/octet-stream",
     });
-    form.set(name, file, basename(rawPath));
+    form.append(name, file, basename(rawPath));
   } else {
-    form.set(name, value);
+    form.append(name, value);
   }
 }
 
@@ -1742,6 +1763,7 @@ async function waitBeforeRetry(
   response: Response | undefined,
   options: RequestOptions,
   attempt: number,
+  signal: AbortSignal | null | undefined,
 ) {
   const retryAfter =
     response && (options.retryAfter || options.retries !== undefined)
@@ -1752,7 +1774,7 @@ async function waitBeforeRetry(
   const jitter = options.retryJitter
     ? Math.floor(exponential * ((Math.random() * options.retryJitter) / 100))
     : 0;
-  await sleep(retryAfter ?? exponential + jitter);
+  await sleep(retryAfter ?? exponential + jitter, undefined, { signal: signal ?? undefined });
 }
 
 function retryAfterMs(value: string | null) {
@@ -1764,26 +1786,61 @@ function retryAfterMs(value: string | null) {
   return undefined;
 }
 
-function sseToNdjson(text: string) {
-  const lines: string[] = [];
-  for (const event of text.split(/\n\n+/)) {
-    const eventName =
-      event
-        .split("\n")
-        .find((line) => line.startsWith("event:"))
-        ?.slice("event:".length)
-        .trim() || "data";
-    const data = event
-      .split("\n")
-      .filter((line) => line.startsWith("data:"))
-      .map((line) => line.slice("data:".length).trimStart())
-      .join("\n");
-    if (data)
-      lines.push(
-        `${JSON.stringify({ event: eventName, data: parseSseData(data), ts: new Date().toISOString() })}\n`,
-      );
+async function* sseToNdjson(response: Response, headerText: string, signal?: AbortSignal) {
+  if (!response.body) {
+    if (headerText) yield headerText;
+    return;
+  }
+  const reader = response.body.getReader();
+  const cancel = () => void reader.cancel(signal?.reason).catch(() => undefined);
+  if (signal?.aborted) cancel();
+  else signal?.addEventListener("abort", cancel, { once: true });
+  const decoder = new TextDecoder();
+  let pending = "";
+  let skipLeadingLf = false;
+  let eventName = "data";
+  let data: string[] = [];
+  try {
+    if (headerText) yield headerText;
+    while (true) {
+      const { value, done } = await reader.read();
+      pending += done ? decoder.decode() : decoder.decode(value, { stream: true });
+      while (true) {
+        if (skipLeadingLf) {
+          if (pending.length === 0) break;
+          if (pending[0] === "\n") pending = pending.slice(1);
+          skipLeadingLf = false;
+        }
+        const boundary = pending.search(/[\r\n]/);
+        if (boundary < 0) break;
+        const line = pending.slice(0, boundary);
+        skipLeadingLf = pending[boundary] === "\r";
+        pending = pending.slice(boundary + 1);
+        if (line === "") {
+          if (data.length > 0)
+            yield `${JSON.stringify({ event: eventName, data: parseSseData(data.join("\n")), ts: new Date().toISOString() })}\n`;
+          eventName = "data";
+          data = [];
+        } else {
+          const separator = line.indexOf(":");
+          const field = separator < 0 ? line : line.slice(0, separator);
+          const raw = separator < 0 ? "" : line.slice(separator + 1);
+          const value = raw.startsWith(" ") ? raw.slice(1) : raw;
+          if (field === "event") eventName = value || "data";
+          else if (field === "data") data.push(value);
+        }
+      }
+      // SSE dispatches only events terminated by a blank line.
+      if (done) break;
+    }
+  } finally {
+    signal?.removeEventListener("abort", cancel);
+    try {
+      await reader.cancel();
+    } finally {
+      reader.releaseLock();
+    }
   }
-  return lines.join("");
 }
 
 function parseSseData(data: string) {