fix(remote-mock): abort stream handle on consumer return

packages/test-support/remote-mock/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/test-support/remote-mock/README.md
-README.md: 48cfeec7c6a49c6bd7017db27d2feda4052858ad
-README.zh.md: ca230762467f0e702622cd5dcbc24e2552679d58
+README.md: c3182b61a67aa6af799fb4fbac85083fbe805e59
+README.zh.md: 8199bc748d96df34de28d225db1289d76be6eed2

packages/test-support/remote-mock/README.md

@@ -73,7 +73,7 @@ mock.streams.fail('session/follow', new Error('gone'))
 await mock.streams.drained('session/follow')
 ```
 
-A failed stream rejects the consumer's next read with the given `Error`. Consumer cancellation (the open `signal`, or an early `return`) ends the iteration without throwing and logs the stream as `cancelled`.
+A failed stream rejects the consumer's next read with the given `Error`. Consumer cancellation (the opening signal or an early iterator `return()`) aborts `StreamHandle.signal`, ends the iteration without throwing, and logs the stream as `cancelled`.
 
 ### Connect a client
 

packages/test-support/remote-mock/README.zh.md

@@ -73,7 +73,7 @@ mock.streams.fail('session/follow', new Error('gone'))
 await mock.streams.drained('session/follow')
 ```
 
-失败的流让消费方的下一次读取以给定的 `Error` reject。消费方取消（打开时的 `signal`，或提前 `return`）结束迭代而不抛错，并把该流记为 `cancelled`。
+失败的流让消费方的下一次读取以给定的 `Error` reject。消费方取消（打开时的 signal 或 iterator 提前 `return()`）会中止 `StreamHandle.signal`、结束迭代而不抛错，并把该流记为 `cancelled`。
 
 ### 接上客户端
 

packages/test-support/remote-mock/src/streams.ts

@@ -10,7 +10,7 @@ export interface StreamHandle {
   end(): void
   /** Fail the consumer's next read with `error` after the queued items drain. */
   fail(error: Error): void
-  /** Consumer cancellation. */
+  /** Aborts when the opening signal aborts or the consumer returns early. */
   readonly signal: AbortSignal
 }
 
@@ -52,9 +52,11 @@ export class MockStream implements StreamHandle, AsyncIterable<unknown> {
   private settled: Settled | undefined
   private waiting: { readonly resolve: (result: IteratorResult<unknown>) => void; readonly reject: (error: unknown) => void } | undefined
   private drainWaiters: (() => void)[] = []
+  private readonly cancellation = new AbortController()
+  readonly signal = this.cancellation.signal
 
   /** Consumer cancellation listener; attached while the stream is open, removed when it settles or cancels. */
-  private readonly onAbort = (): void => { this.cancel() }
+  private readonly onAbort = (): void => { this.cancel(this.sourceSignal.reason) }
 
   /** Items pushed but not yet pulled by the consumer. */
   get queued(): number {
@@ -63,11 +65,11 @@ export class MockStream implements StreamHandle, AsyncIterable<unknown> {
 
   /**
    * @param record - log entry this stream updates.
-   * @param signal - consumer cancellation.
+   * @param sourceSignal - cancellation from the caller that opened the stream.
    */
-  constructor(readonly record: StreamRecord, readonly signal: AbortSignal) {
-    if (signal.aborted) this.cancel()
-    else signal.addEventListener('abort', this.onAbort, { once: true })
+  constructor(readonly record: StreamRecord, private readonly sourceSignal: AbortSignal) {
+    if (sourceSignal.aborted) this.cancel(sourceSignal.reason)
+    else sourceSignal.addEventListener('abort', this.onAbort, { once: true })
   }
 
   push(item: unknown): void {
@@ -148,7 +150,7 @@ export class MockStream implements StreamHandle, AsyncIterable<unknown> {
     if (this.record.state !== 'open') return
     this.record.state = state
     this.settled = settled
-    this.signal.removeEventListener('abort', this.onAbort)
+    this.sourceSignal.removeEventListener('abort', this.onAbort)
     this.wakeDrained()
     // A pending read exists only while the queue is empty: push() resolves it directly instead of queueing.
     const waiting = this.waiting
@@ -157,12 +159,14 @@ export class MockStream implements StreamHandle, AsyncIterable<unknown> {
     void this.finish(settled).then(waiting.resolve, waiting.reject)
   }
 
-  private cancel(): void {
+  private cancel(reason?: unknown): void {
     if (this.record.state === 'open') {
       this.record.state = 'cancelled'
-      this.signal.removeEventListener('abort', this.onAbort)
+      this.sourceSignal.removeEventListener('abort', this.onAbort)
       const waiting = this.waiting
       this.waiting = undefined
+      this.queue.length = 0
+      this.cancellation.abort(reason)
       waiting?.resolve({ value: undefined, done: true })
     }
     // Whether cancelled or left after the producer settled, the consumer pulls nothing more: drop what it left.

packages/test-support/remote-mock/tests/proxy.client.spec.ts

@@ -170,7 +170,7 @@ describe('RemoteMock.remote stream proxies', () => {
     expect(mock.log.streams('session/control').map(entry => entry.state)).toEqual(['ended', 'ended'])
   })
 
-  it('shares stream spies with the carrier, preserves its args, and forwards each caller signal', async () => {
+  it('shares stream spies with the carrier, preserves its args, and reflects each caller cancellation', async () => {
     const signals: AbortSignal[] = []
     const script = vi.fn((_args: readonly unknown[], stream: StreamHandle) => {
       signals.push(stream.signal)
@@ -192,7 +192,8 @@ describe('RemoteMock.remote stream proxies', () => {
     await expect(wire.next()).resolves.toEqual({ value: baseline, done: false })
     expect(control.mock.calls).toEqual([[localController.signal], [request, wireController.signal]])
     expect(script.mock.calls.map(([args]) => args)).toEqual([[], [request]])
-    expect(signals).toEqual([localController.signal, wireController.signal])
+    expect(signals).toHaveLength(2)
+    expect(signals.map(signal => signal.aborted)).toEqual([false, false])
     expect(mock.log.streams('session/control').map(entry => entry.args)).toEqual([[], [request]])
 
     const waiting = local.next()
@@ -200,6 +201,7 @@ describe('RemoteMock.remote stream proxies', () => {
     await expect(waiting).resolves.toEqual({ value: undefined, done: true })
     await wire.return!()
     await mock.streams.drained('session/control')
+    expect(signals.map(signal => signal.aborted)).toEqual([true, true])
     expect(mock.streams.push('session/control', baseline)).toBe(0)
     expect(mock.log.streams('session/control').map(entry => entry.state)).toEqual(['cancelled', 'cancelled'])
   })

packages/test-support/remote-mock/tests/streams.client.spec.ts

@@ -95,6 +95,21 @@ describe('RemoteMock streams', () => {
     expect(mock.log.streams().map(entry => entry.state)).toEqual(['cancelled', 'cancelled', 'cancelled'])
   })
 
+  it('aborts the stream handle signal when the consumer returns', async () => {
+    let signal: AbortSignal | undefined
+    const mock = RemoteMock.create().stream('s/f', (_args, stream) => {
+      signal = stream.signal
+      stream.push('first')
+    })
+    const reader = mock.open('s/f', [], idle())[Symbol.asyncIterator]()
+    await expect(reader.next()).resolves.toEqual({ value: 'first', done: false })
+    await expect(reader.return!()).resolves.toEqual({ value: undefined, done: true })
+    expect({
+      state: mock.log.streams('s/f')[0]?.state,
+      signalAborted: signal?.aborted,
+    }).toEqual({ state: 'cancelled', signalAborted: true })
+  })
+
   it('runs script functions with the open args and fails the stream when they throw or reject', async () => {
     const mock = RemoteMock.create()
       .stream('s/echo', (args, stream) => {