From 3bfa9a9ebb425b05ee80b6cf26567daeef2923d2 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 28 Sep 2026 03:59:12 +0000
Subject: [PATCH 1/5] fix(peer): resend a cancel that may have overtaken its
 request message

When a client request was aborted while its request message was still
being sent (e.g. `encodePeerMessage` awaiting `blob.arrayBuffer()`), the
cancel message could reach the server first. The server ignores cancels
for ids it has not seen, so the handler then ran un-aborted and a
streaming response kept going until the peer closed.

The client now sends the cancel again once the in-flight request message
is sent. The immediate cancel is kept for transports whose `send` only
resolves after the server handled the message, where it is the only way
to abort the running handler. A duplicate cancel is a no-op on the server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TSY5GvoAft9apocjwoSQrk
---
 packages/peer/src/client.test.ts | 89 ++++++++++++++++++++++++++++++-
 packages/peer/src/client.ts      | 21 ++++++++
 packages/peer/tests/peer.test.ts | 91 ++++++++++++++++++++++++++++++++
 3 files changed, 200 insertions(+), 1 deletion(-)

diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts
index b0edc650..3c09e1a7 100644
--- a/packages/peer/src/client.test.ts
+++ b/packages/peer/src/client.test.ts
@@ -253,9 +253,11 @@ describe('clientPeer', () => {
         peer.request(makeRequest({ signal: controller.signal })),
       ).rejects.toThrow('aborted during send')
 
-      expect(send).toHaveBeenCalledTimes(2)
+      // the cancel sent during the request message may overtake it, so it is sent again afterwards
+      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(3))
       const id = (send.mock.calls[0]![0] as PeerRequestMessage).id
       expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
+      expect(send).toHaveBeenNthCalledWith(3, { id, kind: 'cancel' })
     })
 
     it('rejects when send throws', async () => {
@@ -328,9 +330,94 @@ describe('clientPeer', () => {
       await expect(peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))).rejects.toBe(error)
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
       expect(cancel).toHaveBeenCalledWith(error)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel', 'cancel'])
+    })
+
+    it('sends the cancel message again once the in-flight request message is sent', async () => {
+      const controller = new AbortController()
+      const requestSent = promiseWithResolvers<void>()
+      send.mockImplementation(async (message) => {
+        if (message.kind === 'request') {
+          await requestSent.promise
+        }
+      })
+
+      const promise = peer.request(makeRequest({ signal: controller.signal }))
+      const id = await waitForSend()
+      const error = new Error('aborted during send')
+      controller.abort(error)
+
+      // cancel right away, in case the transport only resolves `send` once the server handled the request
+      await expect(promise).rejects.toBe(error)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
+
+      // and again once the request message is sent, in case the first cancel overtook it
+      requestSent.resolve()
+      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(3))
+      expect(send).toHaveBeenNthCalledWith(3, { id, kind: 'cancel' })
+    })
+
+    it('does not send the cancel message again when the in-flight request message fails to send', async () => {
+      const controller = new AbortController()
+      const requestSent = promiseWithResolvers<void>()
+      send.mockImplementation(async (message) => {
+        if (message.kind === 'request') {
+          await requestSent.promise
+        }
+      })
+      const cancel = vi.fn()
+
+      const promise = peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))
+      await waitForSend()
+      const error = new Error('aborted during send')
+      controller.abort(error)
+      await expect(promise).rejects.toBe(error)
+
+      // the server never received the request, so the first cancel is enough
+      requestSent.reject(new Error('transport down'))
+      await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
+      await sleep(1)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
+    })
+
+    it('does not send the cancel message again when aborted after the request message is sent', async () => {
+      const controller = new AbortController()
+      const promise = peer.request(makeRequest({ signal: controller.signal }))
+      await waitForSend()
+      await sleep(1)
+
+      controller.abort(new Error('user abort'))
+      await expect(promise).rejects.toThrow('user abort')
+      await sleep(1)
       expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
     })
 
+    it('silently ignores transport failures when sending the cancel message again', async () => {
+      const controller = new AbortController()
+      const requestSent = promiseWithResolvers<void>()
+      send.mockImplementation(async (message) => {
+        if (message.kind === 'request') {
+          await requestSent.promise
+        }
+        else if (message.kind === 'cancel') {
+          throw new Error('transport down')
+        }
+      })
+      const cancel = vi.fn()
+
+      const promise = peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))
+      await waitForSend()
+      const error = new Error('aborted during send')
+      controller.abort(error)
+      await expect(promise).rejects.toBe(error)
+
+      requestSent.resolve()
+      await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
+      // the failed cancel delivery does not replace the abort reason
+      expect(cancel).toHaveBeenCalledWith(error)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel', 'cancel'])
+    })
+
     it('rejects pending request on server abort', async () => {
       const { id, promise } = await requestAndGetId()
       await peer.message(makeCancelMessage(id))
diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts
index e6a4d12d..114fddbf 100644
--- a/packages/peer/src/client.ts
+++ b/packages/peer/src/client.ts
@@ -15,6 +15,13 @@ interface ClientPeerRequestStateInternal {
   eventStreamTransmitter?: EventStreamTransmitter | undefined
   octetStreamTransmitter?: OctetStreamTransmitter | undefined
   removeAbortListener?: (() => void) | undefined
+  /**
+   * A cancel message sent while the request message is in flight can reach the server first,
+   * and the server ignores cancels for requests it has not seen, so such a cancel is sent again
+   * once the request message is sent.
+   */
+  sendingRequestMessage?: boolean | undefined
+  resendCancel?: boolean | undefined
 }
 
 export class ClientPeer {
@@ -77,6 +84,7 @@ export class ClientPeer {
       }
 
       // PeerRequestMessage must be sent before stream messages
+      state.sendingRequestMessage = true
       await this.send({
         id,
         kind: 'request',
@@ -88,9 +96,16 @@ export class ClientPeer {
         },
         binary: encodedAtomicBody.binary,
       })
+      state.sendingRequestMessage = undefined
 
       // The request can already be settled/cancelled while was in flight
       if (this.requests.get(id) !== state) {
+        if (state.resendCancel) {
+          // the first cancel may have overtaken the request message, otherwise the server ignores this one
+          // a failed cancel delivery must not surface as an unhandled rejection
+          await this.send({ id, kind: 'cancel' }).catch(() => {})
+        }
+
         return
       }
 
@@ -265,6 +280,12 @@ export class ClientPeer {
     state.eventStreamMessageQueue = undefined
     state.octetStreamMessageQueue = undefined
 
+    /**
+     * Still cancel right away even if the request message is in flight: some transports only resolve
+     * `send` once the server has handled the message, and this cancel is what aborts that handling.
+     */
+    state.resendCancel = state.sendingRequestMessage
+
     const promises = [
       this.send({ id, kind: 'cancel' }),
       state.eventStreamTransmitter?.cancel(),
diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts
index 9b3ec843..b1572f97 100644
--- a/packages/peer/tests/peer.test.ts
+++ b/packages/peer/tests/peer.test.ts
@@ -2,6 +2,7 @@ import type { StandardLazyRequest, StandardResponse } from '@standard-server/cor
 import type { ClientPeer, ServerPeer } from '../src'
 import type { ClientPeerSendMessage, ServerPeerSendMessage } from '../src/types'
 import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core'
+import { promiseWithResolvers } from '@standard-server/shared'
 import { ClientPeer as ClientPeerClass, decodePeerMessage, encodePeerMessage, ServerPeer as ServerPeerClass } from '../src'
 
 /**
@@ -289,6 +290,96 @@ describe('peer integration (client <-> server over encoded wire)', () => {
     await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true))
   })
 
+  it('propagates a client abort fired while the request message is still being sent', async () => {
+    const prefix = 'peer:'
+    const wire = {} as { client: ClientPeer, server: ServerPeer }
+    const requestSendStarted = promiseWithResolvers<void>()
+    const requestEncoded = promiseWithResolvers<void>()
+
+    let serverSignal: AbortSignal | undefined
+    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => {
+      serverSignal = request.signal
+      return new Promise(() => {}) // handler never resolves
+    }
+
+    // like `connect()`, but encoding the request message (e.g. awaiting `blob.arrayBuffer()`)
+    // takes longer than encoding the cancel message that follows it
+    wire.client = new ClientPeerClass(async (message) => {
+      if (message.kind === 'request') {
+        requestSendStarted.resolve()
+        await requestEncoded.promise
+      }
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      void wire.server.message(decoded.message as ClientPeerSendMessage, handler).catch(() => {})
+    })
+
+    wire.server = new ServerPeerClass(async (message) => {
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      void wire.client.message(decoded.message as ServerPeerSendMessage)
+    })
+
+    const controller = new AbortController()
+    const promise = wire.client.request({
+      url: '/upload',
+      method: 'POST',
+      headers: {},
+      body: new File(['hello'], 'hello.txt', { type: 'text/plain' }),
+      signal: controller.signal,
+    })
+
+    await requestSendStarted.promise
+    controller.abort(new Error('user navigated away'))
+    await expect(promise).rejects.toThrow('user navigated away')
+
+    requestEncoded.resolve()
+    await vi.waitFor(() => expect(serverSignal).toBeDefined())
+    await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true))
+  })
+
+  it('propagates client aborts when the transport waits for full remote processing before send resolves', async () => {
+    const prefix = 'peer:'
+    const wire = {} as { client: ClientPeer, server: ServerPeer }
+
+    let serverSignal: AbortSignal | undefined
+    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => {
+      serverSignal = request.signal
+      await new Promise(resolve => request.signal!.addEventListener('abort', resolve))
+      return { status: 200, headers: {}, body: 'too late' }
+    }
+
+    // the request `send` stays in flight until the handler settles, so only a cancel sent meanwhile can abort it
+    wire.client = new ClientPeerClass(async (message) => {
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      await wire.server.message(decoded.message as ClientPeerSendMessage, handler)
+    })
+
+    wire.server = new ServerPeerClass(async (message) => {
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      await wire.client.message(decoded.message as ServerPeerSendMessage)
+    })
+
+    const controller = new AbortController()
+    const promise = wire.client.request({ url: '/slow', method: 'GET', headers: {}, signal: controller.signal })
+
+    await vi.waitFor(() => expect(serverSignal).toBeDefined())
+    controller.abort(new Error('user navigated away'))
+
+    await expect(promise).rejects.toThrow('user navigated away')
+    await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true))
+  })
+
   it('carries malicious __proto__ payloads as inert data without polluting prototypes', async () => {
     let serverBody: Record<string, unknown> | undefined
 

From f57c258d171851cf5f03b9299a2a3fbdc1e0a19b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 29 Sep 2026 02:38:45 +0000
Subject: [PATCH 2/5] refactor(peer): simplify the cancel resend and its tests

- Track a single `cancelSent` flag instead of two: `transmitRequest` only
  finds the state gone right after the request send resolves, so the
  in-flight window is already implied there.
- Fire the resent cancel without awaiting it, so releasing the untransmitted
  request body is not delayed by another `send`.
- Tests: give `connect()` `awaitRemote` and `beforeClientSend` options
  instead of copying the wire setup, run the client abort propagation test
  for both transport styles, share the abort-while-sending setup in the
  client tests, and fold a duplicate test into the existing one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TSY5GvoAft9apocjwoSQrk
---
 packages/peer/src/client.test.ts |  85 ++++++-----------
 packages/peer/src/client.ts      |  29 +++---
 packages/peer/tests/peer.test.ts | 156 +++++++++++--------------------
 3 files changed, 91 insertions(+), 179 deletions(-)

diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts
index 3c09e1a7..ac6c044e 100644
--- a/packages/peer/src/client.test.ts
+++ b/packages/peer/src/client.test.ts
@@ -208,6 +208,26 @@ describe('clientPeer', () => {
   })
 
   describe('signal / abort', () => {
+    async function abortWhileSendingRequest(body?: StandardRequest['body']) {
+      const requestSent = promiseWithResolvers<void>()
+      send.mockImplementation(async (message) => {
+        if (message.kind === 'request') {
+          await requestSent.promise
+        }
+      })
+
+      const controller = new AbortController()
+      const { id, promise } = await requestAndGetId(makeRequest({ body, signal: controller.signal }))
+      const error = new Error('aborted during send')
+      controller.abort(error)
+
+      // cancel right away, in case the transport only resolves `send` once the server handled the request
+      await expect(promise).rejects.toBe(error)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
+
+      return { id, error, requestSent }
+    }
+
     it('throws immediately if signal already aborted', async () => {
       const controller = new AbortController()
       controller.abort(new Error('pre-aborted'))
@@ -226,6 +246,7 @@ describe('clientPeer', () => {
       controller.abort(error)
 
       await expect(promise).rejects.toThrow(error)
+      await sleep(1) // the cancel is not sent again
       expect(send).toHaveBeenCalledTimes(2)
       expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
     })
@@ -334,82 +355,28 @@ describe('clientPeer', () => {
     })
 
     it('sends the cancel message again once the in-flight request message is sent', async () => {
-      const controller = new AbortController()
-      const requestSent = promiseWithResolvers<void>()
-      send.mockImplementation(async (message) => {
-        if (message.kind === 'request') {
-          await requestSent.promise
-        }
-      })
+      const { id, requestSent } = await abortWhileSendingRequest()
 
-      const promise = peer.request(makeRequest({ signal: controller.signal }))
-      const id = await waitForSend()
-      const error = new Error('aborted during send')
-      controller.abort(error)
-
-      // cancel right away, in case the transport only resolves `send` once the server handled the request
-      await expect(promise).rejects.toBe(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
-
-      // and again once the request message is sent, in case the first cancel overtook it
+      // the first cancel may have overtaken the request message
       requestSent.resolve()
       await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(3))
       expect(send).toHaveBeenNthCalledWith(3, { id, kind: 'cancel' })
     })
 
     it('does not send the cancel message again when the in-flight request message fails to send', async () => {
-      const controller = new AbortController()
-      const requestSent = promiseWithResolvers<void>()
-      send.mockImplementation(async (message) => {
-        if (message.kind === 'request') {
-          await requestSent.promise
-        }
-      })
       const cancel = vi.fn()
-
-      const promise = peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))
-      await waitForSend()
-      const error = new Error('aborted during send')
-      controller.abort(error)
-      await expect(promise).rejects.toBe(error)
+      const { requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
 
       // the server never received the request, so the first cancel is enough
       requestSent.reject(new Error('transport down'))
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
-      await sleep(1)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
-    })
-
-    it('does not send the cancel message again when aborted after the request message is sent', async () => {
-      const controller = new AbortController()
-      const promise = peer.request(makeRequest({ signal: controller.signal }))
-      await waitForSend()
-      await sleep(1)
-
-      controller.abort(new Error('user abort'))
-      await expect(promise).rejects.toThrow('user abort')
-      await sleep(1)
       expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
     })
 
     it('silently ignores transport failures when sending the cancel message again', async () => {
-      const controller = new AbortController()
-      const requestSent = promiseWithResolvers<void>()
-      send.mockImplementation(async (message) => {
-        if (message.kind === 'request') {
-          await requestSent.promise
-        }
-        else if (message.kind === 'cancel') {
-          throw new Error('transport down')
-        }
-      })
       const cancel = vi.fn()
-
-      const promise = peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))
-      await waitForSend()
-      const error = new Error('aborted during send')
-      controller.abort(error)
-      await expect(promise).rejects.toBe(error)
+      const { error, requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
+      send.mockRejectedValueOnce(new Error('transport down'))
 
       requestSent.resolve()
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts
index 114fddbf..c965147b 100644
--- a/packages/peer/src/client.ts
+++ b/packages/peer/src/client.ts
@@ -15,13 +15,7 @@ interface ClientPeerRequestStateInternal {
   eventStreamTransmitter?: EventStreamTransmitter | undefined
   octetStreamTransmitter?: OctetStreamTransmitter | undefined
   removeAbortListener?: (() => void) | undefined
-  /**
-   * A cancel message sent while the request message is in flight can reach the server first,
-   * and the server ignores cancels for requests it has not seen, so such a cancel is sent again
-   * once the request message is sent.
-   */
-  sendingRequestMessage?: boolean | undefined
-  resendCancel?: boolean | undefined
+  cancelSent?: boolean | undefined
 }
 
 export class ClientPeer {
@@ -84,7 +78,6 @@ export class ClientPeer {
       }
 
       // PeerRequestMessage must be sent before stream messages
-      state.sendingRequestMessage = true
       await this.send({
         id,
         kind: 'request',
@@ -96,14 +89,16 @@ export class ClientPeer {
         },
         binary: encodedAtomicBody.binary,
       })
-      state.sendingRequestMessage = undefined
 
       // The request can already be settled/cancelled while was in flight
       if (this.requests.get(id) !== state) {
-        if (state.resendCancel) {
-          // the first cancel may have overtaken the request message, otherwise the server ignores this one
-          // a failed cancel delivery must not surface as an unhandled rejection
-          await this.send({ id, kind: 'cancel' }).catch(() => {})
+        if (state.cancelSent) {
+          /**
+           * The cancel may have overtaken the request message, and the server ignores
+           * cancels for requests it has not seen, so send it again now that the request is sent.
+           * A failed cancel delivery must not surface as an unhandled rejection.
+           */
+          void this.send({ id, kind: 'cancel' }).catch(() => {})
         }
 
         return
@@ -280,11 +275,9 @@ export class ClientPeer {
     state.eventStreamMessageQueue = undefined
     state.octetStreamMessageQueue = undefined
 
-    /**
-     * Still cancel right away even if the request message is in flight: some transports only resolve
-     * `send` once the server has handled the message, and this cancel is what aborts that handling.
-     */
-    state.resendCancel = state.sendingRequestMessage
+    // sent right away even while the request message is in flight (transmitRequest resends it afterwards):
+    // some transports only resolve `send` once the server has handled the request, which only this cancel can abort
+    state.cancelSent = true
 
     const promises = [
       this.send({ id, kind: 'cancel' }),
diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts
index b1572f97..4d963581 100644
--- a/packages/peer/tests/peer.test.ts
+++ b/packages/peer/tests/peer.test.ts
@@ -1,33 +1,57 @@
 import type { StandardLazyRequest, StandardResponse } from '@standard-server/core'
 import type { ClientPeer, ServerPeer } from '../src'
-import type { ClientPeerSendMessage, ServerPeerSendMessage } from '../src/types'
+import type { ClientPeerSendMessage, PeerMessage, ServerPeerSendMessage } from '../src/types'
 import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core'
 import { promiseWithResolvers } from '@standard-server/shared'
 import { ClientPeer as ClientPeerClass, decodePeerMessage, encodePeerMessage, ServerPeer as ServerPeerClass } from '../src'
 
+interface ConnectOptions {
+  /**
+   * Make each `send` resolve only once the remote side has handled the message.
+   */
+  awaitRemote?: boolean
+  /**
+   * Runs before a client message is encoded, e.g. to hold it back.
+   */
+  beforeClientSend?: (message: ClientPeerSendMessage) => Promise<void>
+}
+
 /**
  * Wires a ClientPeer and a ServerPeer together through the real codec,
  * simulating a full-duplex connection (e.g. a WebSocket) where every
  * message crosses the wire encoded.
  */
-function connect(handler: (request: StandardLazyRequest) => Promise<StandardResponse>): { client: ClientPeer, server: ServerPeer } {
+function connect(
+  handler: (request: StandardLazyRequest) => Promise<StandardResponse>,
+  { awaitRemote = false, beforeClientSend }: ConnectOptions = {},
+): { client: ClientPeer, server: ServerPeer } {
   const prefix = 'peer:'
   const wire = {} as { client: ClientPeer, server: ServerPeer }
 
-  wire.client = new ClientPeerClass(async (message) => {
+  const transfer = async (message: PeerMessage): Promise<PeerMessage> => {
     const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
     if (!decoded.matched) {
       throw new Error('Failed to decode message on the wire')
     }
-    void wire.server.message(decoded.message as ClientPeerSendMessage, handler).catch(() => {})
+    return decoded.message
+  }
+
+  wire.client = new ClientPeerClass(async (message) => {
+    await beforeClientSend?.(message)
+    const handled = wire.server.message(await transfer(message) as ClientPeerSendMessage, handler)
+    if (awaitRemote) {
+      await handled
+    }
+    else {
+      void handled.catch(() => {})
+    }
   })
 
   wire.server = new ServerPeerClass(async (message) => {
-    const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-    if (!decoded.matched) {
-      throw new Error('Failed to decode message on the wire')
+    const handled = wire.client.message(await transfer(message) as ServerPeerSendMessage)
+    if (awaitRemote) {
+      await handled
     }
-    void wire.client.message(decoded.message as ServerPeerSendMessage)
   })
 
   return wire
@@ -54,34 +78,14 @@ describe('peer integration (client <-> server over encoded wire)', () => {
   })
 
   it('completes when the transport waits for full remote processing before send resolves', async () => {
-    const prefix = 'peer:'
-    const wire = {} as { client: ClientPeer, server: ServerPeer }
-
-    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => ({
+    // the response arrives while the request `send` is still in flight
+    const { client } = connect(async request => ({
       status: 200,
       headers: {},
       body: { pong: request.url },
-    })
-
-    // unlike `connect()`, each send awaits the remote side handling the message,
-    // so the response arrives while the request `send` is still in flight
-    wire.client = new ClientPeerClass(async (message) => {
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      await wire.server.message(decoded.message as ClientPeerSendMessage, handler)
-    })
-
-    wire.server = new ServerPeerClass(async (message) => {
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      await wire.client.message(decoded.message as ServerPeerSendMessage)
-    })
+    }), { awaitRemote: true })
 
-    const response = await wire.client.request({ url: '/ping', method: 'GET', headers: {} })
+    const response = await client.request({ url: '/ping', method: 'GET', headers: {} })
     expect(response.status).toBe(200)
     expect(await response.resolveBody()).toEqual({ pong: '/ping' })
   })
@@ -272,13 +276,15 @@ describe('peer integration (client <-> server over encoded wire)', () => {
     expect(chunks).toEqual([1, 2, 3])
   })
 
-  it('propagates client aborts to the server handler signal', async () => {
+  it.each([false, true])('propagates client aborts to the server handler signal (awaitRemote: %s)', async (awaitRemote) => {
     let serverSignal: AbortSignal | undefined
 
+    // with `awaitRemote`, the request `send` stays in flight until the handler settles
     const { client } = connect(async (request) => {
       serverSignal = request.signal
-      return new Promise(() => {}) // handler never resolves
-    })
+      await new Promise(resolve => request.signal!.addEventListener('abort', resolve))
+      return { status: 200, headers: {}, body: 'too late' }
+    }, { awaitRemote })
 
     const controller = new AbortController()
     const promise = client.request({ url: '/slow', method: 'GET', headers: {}, signal: controller.signal })
@@ -291,41 +297,26 @@ describe('peer integration (client <-> server over encoded wire)', () => {
   })
 
   it('propagates a client abort fired while the request message is still being sent', async () => {
-    const prefix = 'peer:'
-    const wire = {} as { client: ClientPeer, server: ServerPeer }
     const requestSendStarted = promiseWithResolvers<void>()
     const requestEncoded = promiseWithResolvers<void>()
-
     let serverSignal: AbortSignal | undefined
-    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => {
-      serverSignal = request.signal
-      return new Promise(() => {}) // handler never resolves
-    }
 
-    // like `connect()`, but encoding the request message (e.g. awaiting `blob.arrayBuffer()`)
+    // encoding the request message (e.g. awaiting `blob.arrayBuffer()`)
     // takes longer than encoding the cancel message that follows it
-    wire.client = new ClientPeerClass(async (message) => {
-      if (message.kind === 'request') {
-        requestSendStarted.resolve()
-        await requestEncoded.promise
-      }
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      void wire.server.message(decoded.message as ClientPeerSendMessage, handler).catch(() => {})
-    })
-
-    wire.server = new ServerPeerClass(async (message) => {
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      void wire.client.message(decoded.message as ServerPeerSendMessage)
+    const { client } = connect(async (request) => {
+      serverSignal = request.signal
+      return new Promise(() => {}) // handler never resolves
+    }, {
+      beforeClientSend: async (message) => {
+        if (message.kind === 'request') {
+          requestSendStarted.resolve()
+          await requestEncoded.promise
+        }
+      },
     })
 
     const controller = new AbortController()
-    const promise = wire.client.request({
+    const promise = client.request({
       url: '/upload',
       method: 'POST',
       headers: {},
@@ -338,46 +329,7 @@ describe('peer integration (client <-> server over encoded wire)', () => {
     await expect(promise).rejects.toThrow('user navigated away')
 
     requestEncoded.resolve()
-    await vi.waitFor(() => expect(serverSignal).toBeDefined())
-    await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true))
-  })
-
-  it('propagates client aborts when the transport waits for full remote processing before send resolves', async () => {
-    const prefix = 'peer:'
-    const wire = {} as { client: ClientPeer, server: ServerPeer }
-
-    let serverSignal: AbortSignal | undefined
-    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => {
-      serverSignal = request.signal
-      await new Promise(resolve => request.signal!.addEventListener('abort', resolve))
-      return { status: 200, headers: {}, body: 'too late' }
-    }
-
-    // the request `send` stays in flight until the handler settles, so only a cancel sent meanwhile can abort it
-    wire.client = new ClientPeerClass(async (message) => {
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      await wire.server.message(decoded.message as ClientPeerSendMessage, handler)
-    })
-
-    wire.server = new ServerPeerClass(async (message) => {
-      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
-      if (!decoded.matched) {
-        throw new Error('Failed to decode message on the wire')
-      }
-      await wire.client.message(decoded.message as ServerPeerSendMessage)
-    })
-
-    const controller = new AbortController()
-    const promise = wire.client.request({ url: '/slow', method: 'GET', headers: {}, signal: controller.signal })
-
-    await vi.waitFor(() => expect(serverSignal).toBeDefined())
-    controller.abort(new Error('user navigated away'))
-
-    await expect(promise).rejects.toThrow('user navigated away')
-    await vi.waitFor(() => expect(serverSignal!.aborted).toBe(true))
+    await vi.waitFor(() => expect(serverSignal?.aborted).toBe(true))
   })
 
   it('carries malicious __proto__ payloads as inert data without polluting prototypes', async () => {

From bdb15b7fc450ca7828907a36cf7c6bbf1dce6ed2 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 29 Sep 2026 02:49:22 +0000
Subject: [PATCH 3/5] refactor(peer): hold the cancel back until the request
 message is sent

Instead of cancelling right away and resending afterwards, track whether
the request message has been sent. `abortById` only sends the cancel once
it has; otherwise `transmitRequest` checks the signal after the request
send and awaits the cancel there. This sends a single cancel, always after
the request, and none at all when the request is aborted before it is sent.

Transports whose `send` only resolves after the server has handled the
request no longer get a cancel while that handling runs, so the abort test
for that transport style is dropped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TSY5GvoAft9apocjwoSQrk
---
 packages/peer/src/client.test.ts | 35 +++++++++++++++-----------------
 packages/peer/src/client.ts      | 24 ++++++++++------------
 packages/peer/tests/peer.test.ts |  8 +++-----
 3 files changed, 30 insertions(+), 37 deletions(-)

diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts
index ac6c044e..90d7edb9 100644
--- a/packages/peer/src/client.test.ts
+++ b/packages/peer/src/client.test.ts
@@ -221,9 +221,10 @@ describe('clientPeer', () => {
       const error = new Error('aborted during send')
       controller.abort(error)
 
-      // cancel right away, in case the transport only resolves `send` once the server handled the request
+      // the request settles right away, but the cancel must not overtake the request message
       await expect(promise).rejects.toBe(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
+      await sleep(1)
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request'])
 
       return { id, error, requestSent }
     }
@@ -246,7 +247,6 @@ describe('clientPeer', () => {
       controller.abort(error)
 
       await expect(promise).rejects.toThrow(error)
-      await sleep(1) // the cancel is not sent again
       expect(send).toHaveBeenCalledTimes(2)
       expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
     })
@@ -260,8 +260,8 @@ describe('clientPeer', () => {
       controller.abort(new Error('aborted during encode'))
 
       await expect(promise).rejects.toThrow('aborted during encode')
-      // the request message is never sent, only the cancel
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['cancel'])
+      // the request message is never sent, so neither is a cancel
+      expect(send).not.toHaveBeenCalled()
     })
 
     it('throws when signal aborted during send', async () => {
@@ -274,11 +274,9 @@ describe('clientPeer', () => {
         peer.request(makeRequest({ signal: controller.signal })),
       ).rejects.toThrow('aborted during send')
 
-      // the cancel sent during the request message may overtake it, so it is sent again afterwards
-      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(3))
+      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2))
       const id = (send.mock.calls[0]![0] as PeerRequestMessage).id
       expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
-      expect(send).toHaveBeenNthCalledWith(3, { id, kind: 'cancel' })
     })
 
     it('rejects when send throws', async () => {
@@ -298,7 +296,7 @@ describe('clientPeer', () => {
       await expect(promise).rejects.toThrow(error)
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
       expect(cancel).toHaveBeenCalledWith(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['cancel'])
+      expect(send).not.toHaveBeenCalled()
     })
 
     it('returns an event-stream request body when the peer is closed while encoding', async () => {
@@ -351,29 +349,28 @@ describe('clientPeer', () => {
       await expect(peer.request(makeRequest({ body: new ReadableStream({ cancel }), signal: controller.signal }))).rejects.toBe(error)
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
       expect(cancel).toHaveBeenCalledWith(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel', 'cancel'])
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
     })
 
-    it('sends the cancel message again once the in-flight request message is sent', async () => {
+    it('sends the cancel message once the in-flight request message is sent', async () => {
       const { id, requestSent } = await abortWhileSendingRequest()
 
-      // the first cancel may have overtaken the request message
       requestSent.resolve()
-      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(3))
-      expect(send).toHaveBeenNthCalledWith(3, { id, kind: 'cancel' })
+      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2))
+      expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
     })
 
-    it('does not send the cancel message again when the in-flight request message fails to send', async () => {
+    it('does not send a cancel message when the in-flight request message fails to send', async () => {
       const cancel = vi.fn()
       const { requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
 
-      // the server never received the request, so the first cancel is enough
+      // the server never received the request, so there is nothing to cancel there
       requestSent.reject(new Error('transport down'))
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request'])
     })
 
-    it('silently ignores transport failures when sending the cancel message again', async () => {
+    it('ignores transport failures when sending the cancel message after the request message', async () => {
       const cancel = vi.fn()
       const { error, requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
       send.mockRejectedValueOnce(new Error('transport down'))
@@ -382,7 +379,7 @@ describe('clientPeer', () => {
       await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
       // the failed cancel delivery does not replace the abort reason
       expect(cancel).toHaveBeenCalledWith(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel', 'cancel'])
+      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
     })
 
     it('rejects pending request on server abort', async () => {
diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts
index c965147b..3bc104b5 100644
--- a/packages/peer/src/client.ts
+++ b/packages/peer/src/client.ts
@@ -15,7 +15,7 @@ interface ClientPeerRequestStateInternal {
   eventStreamTransmitter?: EventStreamTransmitter | undefined
   octetStreamTransmitter?: OctetStreamTransmitter | undefined
   removeAbortListener?: (() => void) | undefined
-  cancelSent?: boolean | undefined
+  requestSent?: boolean | undefined
 }
 
 export class ClientPeer {
@@ -89,16 +89,14 @@ export class ClientPeer {
         },
         binary: encodedAtomicBody.binary,
       })
+      state.requestSent = true
 
       // The request can already be settled/cancelled while was in flight
       if (this.requests.get(id) !== state) {
-        if (state.cancelSent) {
-          /**
-           * The cancel may have overtaken the request message, and the server ignores
-           * cancels for requests it has not seen, so send it again now that the request is sent.
-           * A failed cancel delivery must not surface as an unhandled rejection.
-           */
-          void this.send({ id, kind: 'cancel' }).catch(() => {})
+        if (request.signal?.aborted) {
+          // the cancel was held back until the request message is sent (see abortById)
+          // a failed cancel delivery must not replace the abort reason
+          await this.send({ id, kind: 'cancel' }).catch(() => {})
         }
 
         return
@@ -275,12 +273,12 @@ export class ClientPeer {
     state.eventStreamMessageQueue = undefined
     state.octetStreamMessageQueue = undefined
 
-    // sent right away even while the request message is in flight (transmitRequest resends it afterwards):
-    // some transports only resolve `send` once the server has handled the request, which only this cancel can abort
-    state.cancelSent = true
-
     const promises = [
-      this.send({ id, kind: 'cancel' }),
+      /**
+       * Sent before the request message, a cancel could reach the server first, and the server
+       * ignores cancels for requests it has not seen. transmitRequest sends it once the request is sent.
+       */
+      state.requestSent ? this.send({ id, kind: 'cancel' }) : undefined,
       state.eventStreamTransmitter?.cancel(),
       state.octetStreamTransmitter?.cancel(),
     ]
diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts
index 4d963581..51660a57 100644
--- a/packages/peer/tests/peer.test.ts
+++ b/packages/peer/tests/peer.test.ts
@@ -276,15 +276,13 @@ describe('peer integration (client <-> server over encoded wire)', () => {
     expect(chunks).toEqual([1, 2, 3])
   })
 
-  it.each([false, true])('propagates client aborts to the server handler signal (awaitRemote: %s)', async (awaitRemote) => {
+  it('propagates client aborts to the server handler signal', async () => {
     let serverSignal: AbortSignal | undefined
 
-    // with `awaitRemote`, the request `send` stays in flight until the handler settles
     const { client } = connect(async (request) => {
       serverSignal = request.signal
-      await new Promise(resolve => request.signal!.addEventListener('abort', resolve))
-      return { status: 200, headers: {}, body: 'too late' }
-    }, { awaitRemote })
+      return new Promise(() => {}) // handler never resolves
+    })
 
     const controller = new AbortController()
     const promise = client.request({ url: '/slow', method: 'GET', headers: {}, signal: controller.signal })

From 97057314b304b680c48aeb80cf6ab278c31acf79 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 29 Sep 2026 03:14:39 +0000
Subject: [PATCH 4/5] refactor(peer): trim the held-back cancel change and its
 tests

- Explain why the cancel is held back once, on `requestSent`, instead of
  at both call sites.
- Client tests: drive the abort-while-sending helper from the `send` mock
  and the released body instead of polling, and fold the failed-cancel
  test into the existing abort-during-send body test.
- Integration tests: restore `connect()` and its wiring from main, and
  hold back the real `file.arrayBuffer()` read instead of a custom hook.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TSY5GvoAft9apocjwoSQrk
---
 packages/peer/src/client.test.ts |  47 +++++++-------
 packages/peer/src/client.ts      |   9 ++-
 packages/peer/tests/peer.test.ts | 102 ++++++++++++++-----------------
 3 files changed, 72 insertions(+), 86 deletions(-)

diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts
index 90d7edb9..3ff81e6c 100644
--- a/packages/peer/src/client.test.ts
+++ b/packages/peer/src/client.test.ts
@@ -208,25 +208,30 @@ describe('clientPeer', () => {
   })
 
   describe('signal / abort', () => {
-    async function abortWhileSendingRequest(body?: StandardRequest['body']) {
-      const requestSent = promiseWithResolvers<void>()
+    async function abortWhileSendingRequest() {
+      const requestSending = promiseWithResolvers<PeerRequestMessage>()
+      const releaseRequest = promiseWithResolvers<void>()
+      const bodyCancelled = promiseWithResolvers<void>()
       send.mockImplementation(async (message) => {
         if (message.kind === 'request') {
-          await requestSent.promise
+          requestSending.resolve(message)
+          await releaseRequest.promise
         }
       })
 
       const controller = new AbortController()
-      const { id, promise } = await requestAndGetId(makeRequest({ body, signal: controller.signal }))
+      const body = new ReadableStream({ cancel: () => bodyCancelled.resolve() })
+      const promise = peer.request(makeRequest({ body, signal: controller.signal }))
+      const { id } = await requestSending.promise
       const error = new Error('aborted during send')
       controller.abort(error)
 
       // the request settles right away, but the cancel must not overtake the request message
       await expect(promise).rejects.toBe(error)
-      await sleep(1)
       expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request'])
 
-      return { id, error, requestSent }
+      // the unsent body is released last, after any cancel message
+      return { id, releaseRequest, bodyCancelled: bodyCancelled.promise }
     }
 
     it('throws immediately if signal already aborted', async () => {
@@ -343,6 +348,10 @@ describe('clientPeer', () => {
         if (message.kind === 'request') {
           controller.abort(error)
         }
+        else {
+          // a failed cancel delivery must not replace the abort reason
+          throw new Error('transport down')
+        }
       })
       const cancel = vi.fn()
 
@@ -353,35 +362,23 @@ describe('clientPeer', () => {
     })
 
     it('sends the cancel message once the in-flight request message is sent', async () => {
-      const { id, requestSent } = await abortWhileSendingRequest()
+      const { id, releaseRequest, bodyCancelled } = await abortWhileSendingRequest()
 
-      requestSent.resolve()
-      await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2))
+      releaseRequest.resolve()
+      await bodyCancelled
+      expect(send).toHaveBeenCalledTimes(2)
       expect(send).toHaveBeenNthCalledWith(2, { id, kind: 'cancel' })
     })
 
     it('does not send a cancel message when the in-flight request message fails to send', async () => {
-      const cancel = vi.fn()
-      const { requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
+      const { releaseRequest, bodyCancelled } = await abortWhileSendingRequest()
 
       // the server never received the request, so there is nothing to cancel there
-      requestSent.reject(new Error('transport down'))
-      await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
+      releaseRequest.reject(new Error('transport down'))
+      await bodyCancelled
       expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request'])
     })
 
-    it('ignores transport failures when sending the cancel message after the request message', async () => {
-      const cancel = vi.fn()
-      const { error, requestSent } = await abortWhileSendingRequest(new ReadableStream({ cancel }))
-      send.mockRejectedValueOnce(new Error('transport down'))
-
-      requestSent.resolve()
-      await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce())
-      // the failed cancel delivery does not replace the abort reason
-      expect(cancel).toHaveBeenCalledWith(error)
-      expect(send.mock.calls.map(([m]) => m.kind)).toEqual(['request', 'cancel'])
-    })
-
     it('rejects pending request on server abort', async () => {
       const { id, promise } = await requestAndGetId()
       await peer.message(makeCancelMessage(id))
diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts
index 3bc104b5..053f0283 100644
--- a/packages/peer/src/client.ts
+++ b/packages/peer/src/client.ts
@@ -15,6 +15,10 @@ interface ClientPeerRequestStateInternal {
   eventStreamTransmitter?: EventStreamTransmitter | undefined
   octetStreamTransmitter?: OctetStreamTransmitter | undefined
   removeAbortListener?: (() => void) | undefined
+  /**
+   * A cancel must not overtake the request message (the server ignores cancels for unknown ids),
+   * so until the request message is sent, transmitRequest sends the cancel instead of abortById.
+   */
   requestSent?: boolean | undefined
 }
 
@@ -94,7 +98,6 @@ export class ClientPeer {
       // The request can already be settled/cancelled while was in flight
       if (this.requests.get(id) !== state) {
         if (request.signal?.aborted) {
-          // the cancel was held back until the request message is sent (see abortById)
           // a failed cancel delivery must not replace the abort reason
           await this.send({ id, kind: 'cancel' }).catch(() => {})
         }
@@ -274,10 +277,6 @@ export class ClientPeer {
     state.octetStreamMessageQueue = undefined
 
     const promises = [
-      /**
-       * Sent before the request message, a cancel could reach the server first, and the server
-       * ignores cancels for requests it has not seen. transmitRequest sends it once the request is sent.
-       */
       state.requestSent ? this.send({ id, kind: 'cancel' }) : undefined,
       state.eventStreamTransmitter?.cancel(),
       state.octetStreamTransmitter?.cancel(),
diff --git a/packages/peer/tests/peer.test.ts b/packages/peer/tests/peer.test.ts
index 51660a57..e6823fc2 100644
--- a/packages/peer/tests/peer.test.ts
+++ b/packages/peer/tests/peer.test.ts
@@ -1,57 +1,33 @@
 import type { StandardLazyRequest, StandardResponse } from '@standard-server/core'
 import type { ClientPeer, ServerPeer } from '../src'
-import type { ClientPeerSendMessage, PeerMessage, ServerPeerSendMessage } from '../src/types'
+import type { ClientPeerSendMessage, ServerPeerSendMessage } from '../src/types'
 import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core'
 import { promiseWithResolvers } from '@standard-server/shared'
 import { ClientPeer as ClientPeerClass, decodePeerMessage, encodePeerMessage, ServerPeer as ServerPeerClass } from '../src'
 
-interface ConnectOptions {
-  /**
-   * Make each `send` resolve only once the remote side has handled the message.
-   */
-  awaitRemote?: boolean
-  /**
-   * Runs before a client message is encoded, e.g. to hold it back.
-   */
-  beforeClientSend?: (message: ClientPeerSendMessage) => Promise<void>
-}
-
 /**
  * Wires a ClientPeer and a ServerPeer together through the real codec,
  * simulating a full-duplex connection (e.g. a WebSocket) where every
  * message crosses the wire encoded.
  */
-function connect(
-  handler: (request: StandardLazyRequest) => Promise<StandardResponse>,
-  { awaitRemote = false, beforeClientSend }: ConnectOptions = {},
-): { client: ClientPeer, server: ServerPeer } {
+function connect(handler: (request: StandardLazyRequest) => Promise<StandardResponse>): { client: ClientPeer, server: ServerPeer } {
   const prefix = 'peer:'
   const wire = {} as { client: ClientPeer, server: ServerPeer }
 
-  const transfer = async (message: PeerMessage): Promise<PeerMessage> => {
+  wire.client = new ClientPeerClass(async (message) => {
     const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
     if (!decoded.matched) {
       throw new Error('Failed to decode message on the wire')
     }
-    return decoded.message
-  }
-
-  wire.client = new ClientPeerClass(async (message) => {
-    await beforeClientSend?.(message)
-    const handled = wire.server.message(await transfer(message) as ClientPeerSendMessage, handler)
-    if (awaitRemote) {
-      await handled
-    }
-    else {
-      void handled.catch(() => {})
-    }
+    void wire.server.message(decoded.message as ClientPeerSendMessage, handler).catch(() => {})
   })
 
   wire.server = new ServerPeerClass(async (message) => {
-    const handled = wire.client.message(await transfer(message) as ServerPeerSendMessage)
-    if (awaitRemote) {
-      await handled
+    const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+    if (!decoded.matched) {
+      throw new Error('Failed to decode message on the wire')
     }
+    void wire.client.message(decoded.message as ServerPeerSendMessage)
   })
 
   return wire
@@ -78,14 +54,34 @@ describe('peer integration (client <-> server over encoded wire)', () => {
   })
 
   it('completes when the transport waits for full remote processing before send resolves', async () => {
-    // the response arrives while the request `send` is still in flight
-    const { client } = connect(async request => ({
+    const prefix = 'peer:'
+    const wire = {} as { client: ClientPeer, server: ServerPeer }
+
+    const handler = async (request: StandardLazyRequest): Promise<StandardResponse> => ({
       status: 200,
       headers: {},
       body: { pong: request.url },
-    }), { awaitRemote: true })
+    })
 
-    const response = await client.request({ url: '/ping', method: 'GET', headers: {} })
+    // unlike `connect()`, each send awaits the remote side handling the message,
+    // so the response arrives while the request `send` is still in flight
+    wire.client = new ClientPeerClass(async (message) => {
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      await wire.server.message(decoded.message as ClientPeerSendMessage, handler)
+    })
+
+    wire.server = new ServerPeerClass(async (message) => {
+      const decoded = decodePeerMessage(await encodePeerMessage(message, { prefix }), { prefix })
+      if (!decoded.matched) {
+        throw new Error('Failed to decode message on the wire')
+      }
+      await wire.client.message(decoded.message as ServerPeerSendMessage)
+    })
+
+    const response = await wire.client.request({ url: '/ping', method: 'GET', headers: {} })
     expect(response.status).toBe(200)
     expect(await response.resolveBody()).toEqual({ pong: '/ping' })
   })
@@ -295,38 +291,32 @@ describe('peer integration (client <-> server over encoded wire)', () => {
   })
 
   it('propagates a client abort fired while the request message is still being sent', async () => {
-    const requestSendStarted = promiseWithResolvers<void>()
-    const requestEncoded = promiseWithResolvers<void>()
+    const encodeStarted = promiseWithResolvers<void>()
+    const releaseEncode = promiseWithResolvers<void>()
     let serverSignal: AbortSignal | undefined
 
-    // encoding the request message (e.g. awaiting `blob.arrayBuffer()`)
-    // takes longer than encoding the cancel message that follows it
     const { client } = connect(async (request) => {
       serverSignal = request.signal
       return new Promise(() => {}) // handler never resolves
-    }, {
-      beforeClientSend: async (message) => {
-        if (message.kind === 'request') {
-          requestSendStarted.resolve()
-          await requestEncoded.promise
-        }
-      },
     })
 
-    const controller = new AbortController()
-    const promise = client.request({
-      url: '/upload',
-      method: 'POST',
-      headers: {},
-      body: new File(['hello'], 'hello.txt', { type: 'text/plain' }),
-      signal: controller.signal,
+    // encoding the request message awaits `file.arrayBuffer()`, so the cancel message could overtake it
+    const file = new File(['hello'], 'hello.txt', { type: 'text/plain' })
+    const arrayBuffer = file.arrayBuffer.bind(file)
+    vi.spyOn(file, 'arrayBuffer').mockImplementation(async () => {
+      encodeStarted.resolve()
+      await releaseEncode.promise
+      return arrayBuffer()
     })
 
-    await requestSendStarted.promise
+    const controller = new AbortController()
+    const promise = client.request({ url: '/upload', method: 'POST', headers: {}, body: file, signal: controller.signal })
+
+    await encodeStarted.promise
     controller.abort(new Error('user navigated away'))
     await expect(promise).rejects.toThrow('user navigated away')
 
-    requestEncoded.resolve()
+    releaseEncode.resolve()
     await vi.waitFor(() => expect(serverSignal?.aborted).toBe(true))
   })
 

From caa86c46367adfc33ae9b646fb86474943d28666 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 29 Sep 2026 03:18:49 +0000
Subject: [PATCH 5/5] refactor(peer): let a failed post-send cancel reach
 transmitRequest's catch

Drop the `.catch(() => {})` on the cancel sent after the request message.
A failed delivery is already handled by the surrounding try/catch, which
closes the request and cancels the unsent body with that error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TSY5GvoAft9apocjwoSQrk
---
 packages/peer/src/client.test.ts | 4 ----
 packages/peer/src/client.ts      | 3 +--
 2 files changed, 1 insertion(+), 6 deletions(-)

diff --git a/packages/peer/src/client.test.ts b/packages/peer/src/client.test.ts
index 3ff81e6c..1c710e95 100644
--- a/packages/peer/src/client.test.ts
+++ b/packages/peer/src/client.test.ts
@@ -348,10 +348,6 @@ describe('clientPeer', () => {
         if (message.kind === 'request') {
           controller.abort(error)
         }
-        else {
-          // a failed cancel delivery must not replace the abort reason
-          throw new Error('transport down')
-        }
       })
       const cancel = vi.fn()
 
diff --git a/packages/peer/src/client.ts b/packages/peer/src/client.ts
index 053f0283..00e2c357 100644
--- a/packages/peer/src/client.ts
+++ b/packages/peer/src/client.ts
@@ -98,8 +98,7 @@ export class ClientPeer {
       // The request can already be settled/cancelled while was in flight
       if (this.requests.get(id) !== state) {
         if (request.signal?.aborted) {
-          // a failed cancel delivery must not replace the abort reason
-          await this.send({ id, kind: 'cancel' }).catch(() => {})
+          await this.send({ id, kind: 'cancel' })
         }
 
         return