From c95614300797dbfe81dc210a0f43a74ecc0e0250 Mon Sep 17 00:00:00 2001
From: Mablr <59505383+mablr@users.noreply.github.com>
Date: Tue, 8 Sep 2026 15:55:54 +0200
Subject: [PATCH 1/3] fix(request): correct SSE, retry, multipart, and HEAD
 handling

Decode SSE into NDJSON incrementally with LF, CRLF and CR framing, preserving
UTF-8 across chunks and cancelling the response when output fails. Await writes to injected output streams.

Make retry backoff respect the request deadline, preserve repeated multipart
fields and files, and route HEAD output through the common writer so -o creates
parent directories and does not also print headers to stdout.
---
 .changelog/request-io.md |   5 +
 src/commands/request.ts  | 118 ++++++++++++++------
 test/request-io.test.ts  | 226 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 319 insertions(+), 30 deletions(-)
 create mode 100644 .changelog/request-io.md
 create mode 100644 test/request-io.test.ts

diff --git a/.changelog/request-io.md b/.changelog/request-io.md
new file mode 100644
index 0000000..04c78ee
--- /dev/null
+++ b/.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.
diff --git a/src/commands/request.ts b/src/commands/request.ts
index da0cf79..c19b53a 100644
--- a/src/commands/request.ts
+++ b/src/commands/request.ts
@@ -451,14 +451,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 +1451,31 @@ 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 === process.stdout) {
+      await pipeline(
+        (options) => sseToNdjson(response, headerText, options?.signal),
+        process.stdout,
+        { end: false },
+      );
+    } else {
+      for await (const chunk of sseToNdjson(response, headerText)) await write(stdout, chunk);
+    }
     return;
   }
 
@@ -1720,9 +1742,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 +1764,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 +1775,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 +1787,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) {
diff --git a/test/request-io.test.ts b/test/request-io.test.ts
new file mode 100644
index 0000000..5a4a03e
--- /dev/null
+++ b/test/request-io.test.ts
@@ -0,0 +1,226 @@
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+import { createServer, type ServerResponse, type IncomingMessage } from "node:http";
+import { readFile, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+import { afterEach, expect, it } from "vitest";
+import { executeRequest, parseRequestArgs, runRequest } from "../src/commands/request.js";
+import { useTempHome } from "./helpers.js";
+
+const execFileAsync = promisify(execFile);
+const cleanup: (() => Promise<void>)[] = [];
+afterEach(async () => {
+  await Promise.all(cleanup.splice(0).map((close) => close()));
+});
+async function server(handler: (req: IncomingMessage, res: ServerResponse) => void) {
+  const instance = createServer(handler);
+  await new Promise<void>((resolve) => instance.listen(0, "127.0.0.1", resolve));
+  cleanup.push(
+    () =>
+      new Promise<void>((resolve, reject) => {
+        instance.closeAllConnections();
+        instance.close((error) => (error ? reject(error) : resolve()));
+      }),
+  );
+  const address = instance.address();
+  if (!address || typeof address === "string") throw new Error("Missing server address");
+  return `http://127.0.0.1:${address.port}`;
+}
+function capture() {
+  let text = "";
+  return {
+    write(chunk: string | Uint8Array) {
+      text += Buffer.from(chunk).toString();
+      return true;
+    },
+    text: () => text,
+  };
+}
+
+it.each(["\n", "\r\n", "\r"])(
+  "streams SSE with %j framing before EOF, preserving split UTF-8",
+  async (newline) => {
+    let finish: (() => void) | undefined;
+    const url = await server((_req, res) => {
+      res.setHeader("content-type", "text/event-stream");
+      const bytes = Buffer.from(`event: token${newline}data: "café"${newline}${newline}`);
+      const split = bytes.indexOf(Buffer.from("é")) + 1;
+      res.write(bytes.subarray(0, split));
+      setImmediate(() => {
+        res.write(bytes.subarray(split));
+      });
+      finish = () =>
+        res.end(`data: first${newline}data:  second${newline}${newline}data: unfinished`);
+    });
+    const stdout = capture();
+    const request = runRequest(["--sse-json", url], { stdout });
+    try {
+      await expect.poll(() => stdout.text(), { timeout: 2000 }).toContain('"data":"café"');
+    } finally {
+      finish?.();
+    }
+    await request;
+    const events = stdout
+      .text()
+      .trim()
+      .split("\n")
+      .map((line) => JSON.parse(line));
+    expect(events).toMatchObject([
+      { event: "token", data: "café" },
+      { event: "data", data: "first\n second" },
+    ]);
+  },
+);
+
+it("writes SSE JSON to a nested output file without stdout", async () => {
+  const home = await useTempHome();
+  const path = join(home, "new", "events.jsonl");
+  const url = await server((_req, res) => res.end("data: {}\r\n\r\ndata:\r\n\r\n"));
+  const stdout = capture();
+  await runRequest(["--sse-json", "-o", path, url], { stdout });
+  expect(stdout.text()).toBe("");
+  expect(
+    (await readFile(path, "utf8"))
+      .trim()
+      .split("\n")
+      .map((line) => JSON.parse(line).data),
+  ).toEqual([{}, ""]);
+});
+
+it("cancels an open SSE response when its output stream fails", async () => {
+  const outputPath = await useTempHome();
+  let upstreamClosed = false;
+  const url = await server((_req, res) => {
+    res.writeHead(200, { "content-type": "text/event-stream" });
+    res.flushHeaders();
+    res.on("close", () => {
+      upstreamClosed = true;
+    });
+  });
+  await expect(runRequest(["--sse-json", "-o", outputPath, url])).rejects.toThrow();
+  await expect.poll(() => upstreamClosed).toBe(true);
+});
+
+it("cancels an open SSE response when creating its output directory fails", async () => {
+  const home = await useTempHome();
+  const parent = join(home, "file");
+  await writeFile(parent, "not a directory");
+  let upstreamClosed = false;
+  const url = await server((_req, res) => {
+    res.writeHead(200, { "content-type": "text/event-stream" });
+    res.flushHeaders();
+    res.on("close", () => {
+      upstreamClosed = true;
+    });
+  });
+  await expect(
+    runRequest(["--sse-json", "-o", join(parent, "events.jsonl"), url]),
+  ).rejects.toThrow();
+  await expect.poll(() => upstreamClosed).toBe(true);
+});
+
+it("aborts Retry-After backoff at the request deadline without another request", async () => {
+  let requests = 0;
+  const url = await server((_req, res) => {
+    requests++;
+    res.writeHead(503, { "retry-after": "5" });
+    res.end("busy");
+  });
+  const started = Date.now();
+  await expect(
+    executeRequest({ ...parseRequestArgs([url]), maxTime: 0.1, retries: 2 }),
+  ).rejects.toMatchObject({ code: "E_NETWORK" });
+  expect(Date.now() - started).toBeLessThan(1500);
+  expect(requests).toBe(1);
+});
+
+it("preserves repeated multipart fields and files in order", async () => {
+  const home = await useTempHome();
+  const first = join(home, "first.txt");
+  const second = join(home, "second.txt");
+  await writeFile(first, "file one");
+  await writeFile(second, "file two");
+  let received = "";
+  const url = await server((req, res) => {
+    req.on("data", (chunk) => {
+      received += chunk;
+    });
+    req.on("end", () => res.end("ok"));
+  });
+  await runRequest(
+    ["-F", "tag=alpha", "-F", "tag=beta", "-F", `file=@${first}`, "-F", `file=@${second}`, url],
+    { stdout: capture() },
+  );
+  expect(received.match(/name="tag"/g)).toHaveLength(2);
+  expect(received.match(/name="file"/g)).toHaveLength(2);
+  expect(received.indexOf("alpha")).toBeLessThan(received.indexOf("beta"));
+  expect(received.indexOf("file one")).toBeLessThan(received.indexOf("file two"));
+});
+
+it("routes HEAD headers exclusively to -o and creates parent directories", async () => {
+  const home = await useTempHome();
+  const path = join(home, "missing", "headers.txt");
+  const url = await server((req, res) => {
+    expect(req.method).toBe("HEAD");
+    res.setHeader("x-test", "head");
+    res.end();
+  });
+  const stdout = capture();
+  const result = await execFileAsync(process.execPath, [
+    "--import",
+    "tsx",
+    "src/request-cli.ts",
+    "-I",
+    "-o",
+    path,
+    url,
+  ]);
+  expect(result.stdout).toBe("");
+  expect(stdout.text()).toBe("");
+  expect(await readFile(path, "utf8")).toContain("x-test: head");
+});
+
+it("handles a CRLF separator split between network writes", async () => {
+  let sendRest: (() => void) | undefined;
+  const url = await server((_req, res) => {
+    res.write("data: 1\r\n\r");
+    sendRest = () => res.end("\ndata: 2\r\n\r\n");
+  });
+  const stdout = capture();
+  const request = runRequest(["--sse-json", url], { stdout });
+  await expect.poll(() => sendRest).toBeTypeOf("function");
+  await expect.poll(() => stdout.text()).toContain('"data":1');
+  sendRest!();
+  await request;
+  expect(
+    stdout
+      .text()
+      .trim()
+      .split("\n")
+      .map((line) => JSON.parse(line).data),
+  ).toEqual([1, 2]);
+});
+
+it("honors fractional CLI timeout while waiting to retry", async () => {
+  let requests = 0;
+  const url = await server((_req, res) => {
+    requests++;
+    res.writeHead(503, { "retry-after": "5" });
+    res.end("busy");
+  });
+  const started = Date.now();
+  await expect(
+    execFileAsync(process.execPath, [
+      "--import",
+      "tsx",
+      "src/request-cli.ts",
+      "-m",
+      "0.1",
+      "--retries",
+      "2",
+      url,
+    ]),
+  ).rejects.toMatchObject({ code: 3 });
+  expect(Date.now() - started).toBeLessThan(2500);
+  expect(requests).toBe(1);
+});

From 340f88ee7770e11063817af27a43c9de9d50c68f Mon Sep 17 00:00:00 2001
From: Mablr <59505383+mablr@users.noreply.github.com>
Date: Tue, 8 Sep 2026 18:50:03 +0200
Subject: [PATCH 2/3] fix(request): propagate injected SSE output failures

---
 src/commands/request.ts | 13 ++++++-------
 test/request-io.test.ts | 38 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 44 insertions(+), 7 deletions(-)

diff --git a/src/commands/request.ts b/src/commands/request.ts
index c19b53a..28031e4 100644
--- a/src/commands/request.ts
+++ b/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";
 
@@ -1467,14 +1468,12 @@ async function writeResponseBody(
         (options) => sseToNdjson(response, headerText, options?.signal),
         createWriteStream(outputPath),
       );
-    } else if (stdout === process.stdout) {
-      await pipeline(
-        (options) => sseToNdjson(response, headerText, options?.signal),
-        process.stdout,
-        { end: false },
-      );
+    } 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)) await write(stdout, chunk);
+      for await (const chunk of sseToNdjson(response, headerText)) write(stdout, chunk);
     }
     return;
   }
diff --git a/test/request-io.test.ts b/test/request-io.test.ts
index 5a4a03e..7d36ce8 100644
--- a/test/request-io.test.ts
+++ b/test/request-io.test.ts
@@ -1,4 +1,5 @@
 import { execFile } from "node:child_process";
+import { Writable } from "node:stream";
 import { promisify } from "node:util";
 import { createServer, type ServerResponse, type IncomingMessage } from "node:http";
 import { readFile, writeFile } from "node:fs/promises";
@@ -101,6 +102,43 @@ it("cancels an open SSE response when its output stream fails", async () => {
   await expect.poll(() => upstreamClosed).toBe(true);
 });
 
+it("cancels an open SSE response when an injected Writable fails asynchronously", async () => {
+  let upstreamClosed = false;
+  let finish: (() => void) | undefined;
+  const url = await server((_req, res) => {
+    res.writeHead(200, { "content-type": "text/event-stream" });
+    res.write("data: 1\n\n");
+    res.on("close", () => {
+      upstreamClosed = true;
+    });
+    finish = () => res.end();
+  });
+  const failure = new Error("output failed");
+  const stdout = new Writable({
+    write(_chunk, _encoding, callback) {
+      setImmediate(() => callback(failure));
+    },
+  });
+  // Keep the pre-fix stream error from becoming an uncaught exception.
+  stdout.on("error", () => {});
+  let outcome: unknown;
+  const request = runRequest(["--sse-json", url], { stdout }).then(
+    () => {
+      outcome = "resolved";
+    },
+    (error: unknown) => {
+      outcome = error;
+    },
+  );
+  try {
+    await expect.poll(() => outcome).toBe(failure);
+    await expect.poll(() => upstreamClosed).toBe(true);
+  } finally {
+    finish?.();
+    await request;
+  }
+});
+
 it("cancels an open SSE response when creating its output directory fails", async () => {
   const home = await useTempHome();
   const parent = join(home, "file");

From e593698bc9e11b312b0e6a476775c2a725f4babe Mon Sep 17 00:00:00 2001
From: Brendan Ryan <1572504+brendanjryan@users.noreply.github.com>
Date: Wed, 9 Sep 2026 17:05:54 +0000
Subject: [PATCH 3/3] test(request): exclude CLI startup from timeout assertion

Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com>
---
 .../friction.md                               | 24 +++++++++++++++++++
 test/request-io.test.ts                       |  6 +++--
 2 files changed, 28 insertions(+), 2 deletions(-)
 create mode 100644 .agents/friction-log/20260909170520-cli-timeout-test/friction.md

diff --git a/.agents/friction-log/20260909170520-cli-timeout-test/friction.md b/.agents/friction-log/20260909170520-cli-timeout-test/friction.md
new file mode 100644
index 0000000..47f7c0a
--- /dev/null
+++ b/.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.
diff --git a/test/request-io.test.ts b/test/request-io.test.ts
index 7d36ce8..2e1ef82 100644
--- a/test/request-io.test.ts
+++ b/test/request-io.test.ts
@@ -241,12 +241,14 @@ it("handles a CRLF separator split between network writes", async () => {
 
 it("honors fractional CLI timeout while waiting to retry", async () => {
   let requests = 0;
+  let requestStarted = 0;
   const url = await server((_req, res) => {
     requests++;
+    // Exclude Node/tsx startup from the request deadline assertion.
+    requestStarted = performance.now();
     res.writeHead(503, { "retry-after": "5" });
     res.end("busy");
   });
-  const started = Date.now();
   await expect(
     execFileAsync(process.execPath, [
       "--import",
@@ -259,6 +261,6 @@ it("honors fractional CLI timeout while waiting to retry", async () => {
       url,
     ]),
   ).rejects.toMatchObject({ code: 3 });
-  expect(Date.now() - started).toBeLessThan(2500);
   expect(requests).toBe(1);
+  expect(performance.now() - requestStarted).toBeLessThan(2500);
 });