Merge pull request #5041 from deepseek-harness/fix-gateway-mem

fix(gateway): bound stream cancellation retention

.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.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 .agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md
-2026-09-19-remote-duplex-stream.md: 0f1f6a575f6871d37822c5829fe4531261b8e95b
-2026-09-19-remote-duplex-stream.zh.md: d3982aa433aa3ce84b945bb5831cef8064472853
+2026-09-19-remote-duplex-stream.md: ea8e8cebcc8285c97ebdb70f4e37805d3ce8bfe9
+2026-09-19-remote-duplex-stream.zh.md: 2e119bf054e20d076e22758178b5e0168092be18

.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md

@@ -276,6 +276,12 @@ Parsing rules: `item` has exactly the keys `type` and `streamId` plus an optiona
 | `item` received after `end` | Any | The stream fails with a `gateway/protocol` error frame and aborts; the socket stays open |
 | `item` / `end` with an unknown `streamId` | Any | Ignored, like `cancel`: after the Host ends a stream and deletes its id, uplink frames still in flight from the Client are normal and must not take down the other streams on the same socket |
 
+### Cancellation wait lifetime
+
+`cancellableStream`, `UplinkDecoder.next()`, and the Client uplink pump each create a cancellation or stop promise per read. `Promise.race` does not remove the losing input's reaction when a read completes: reusing a stream-wide pending promise retains completed races and their item results until that promise settles or becomes unreachable. Read-local promises keep retained cancellation state independent of the number of delivered items.
+
+The Host downlink clears its current reject callback after each read and checks cancellation before reading again. The uplink decoder retains only pending reads: abort rejects them, while normal close resolves all of them with `done`, including overlapping reads. The Client pump drops its stop promise after each read and does not start another read after stopping. Source iterators retain their existing release ownership; closing the Host uplink still precedes returning the method iterator.
+
 ### Ordering guarantees
 
 Frames on one socket arrive in total order: a logical stream's `open`, `item`…, `end` reach the Host in send order, and uplink items enter `uplink()` in the order of `send`. There is no cross-direction ordering guarantee between uplink and downlink; a protocol that needs request-response pairing carries its own sequence numbers.
@@ -538,6 +544,10 @@ for await (const reply of stream) replies.push(reply)   // ['> a', '> b']
 
 ## Alternatives considered
 
+**A shared cancellation controller for all three reading loops.** Each loop already has an owner with distinct close behavior. Read-local promises bound retained state without moving those responsibilities into a new abstraction.
+
+**Requiring source iterators to cancel their own `next()`.** Gateway accepts arbitrary iterables, whose pending reads need not respond to cancellation or `return()`. Gateway must still wake its own waiters independently of source cleanup.
+
 **A separate `mode: 'duplex'` plus a reserved parameter `uplink: AsyncIterable<In>` alongside `signal`.** Special logic everywhere: the analyzer must recognize a second reserved parameter and its position rules, the generated signature gains a parameter, load time must validate `duplex ⟺ uplink`, and because the mux does not know the method mode, the Gateway must reject uplink frames on behalf of `'stream'` methods. Duplex is an inherent property of the transport-layer stream, not a mode of the method: the carrier already has `streamId`, `open`, and `cancel`; the uplink adds just two more frames, which any stream can receive and which are simply not sent when unused.
 
 **Declaring the uplink type in a decorator option, `@Remote({ mode: 'stream', uplink: 'JobInputFrame' })`.** A string reference to a type name is not checked by the compiler and breaks on rename. The return-type alias puts both types in one compiler-visible place.
@@ -572,6 +582,8 @@ for await (const reply of stream) replies.push(reply)   // ['> a', '> b']
 
 ## Testing
 
+Retention cases in `gateway-stream.host.spec.ts` and `gateway.client.spec.ts` count live iterator results after a full GC while the streams remain open. Lifecycle cases cover normal close waking multiple pending uplink reads, cancellation between reads, synchronous abort followed by a source exception, and termination during a Client send.
+
 - `packages/api/gateway/tests/echo-stream.host.spec.ts`: the same echo method passes through the WebSocket carrier and through the in-process carrier.
 - Gateway Host: exact-key validation in frame parsing; `UplinkInbox` buffering during opening, `item` after `end`, overflow, ignored unknown `streamId`, and cleanup on socket closure; `UplinkDecoder` delivery with and without a codec and abort on decode failure; `uplink()` taken only once; the `extend({ invocation })` view and the origin of `peer`; `cancellableStream` closing the uplink before `return()`.
 - Gateway Client: the handle's `send` (JSON validation, queued before open, throws after termination), idempotent `end`, `dispose` driving the `cancel` frame and ending later reads; the pump's three exits; the pump stopping as soon as a terminal frame arrives.

.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.zh.md

@@ -276,6 +276,12 @@ export type RemoteStreamServerMessage =
 | `end` 之后又收到 `item` | 任意 | 该流以 `gateway/protocol` 错误帧失败并中止；socket 不关闭 |
 | 未知 `streamId` 的 `item` / `end` | 任意 | 忽略，与 `cancel` 相同：Host 结束流并删除 id 后，客户端在途的上行帧属正常现象，不能连累同一 socket 上的其他流 |
 
+### 取消等待的生命周期
+
+`cancellableStream`、`UplinkDecoder.next()` 和 Client 上行泵各自为单次读取创建取消或停止 Promise。读取完成后，`Promise.race` 不会移除另一输入上的 reaction：复用流级未完成 Promise 会保留已完成的竞争及其读取结果，直到该 Promise 完成或不可达。单次读取的 Promise 使取消状态的保留量不随已交付项数增长。
+
+Host 下行在每次读取后清除当前拒绝回调，并在再次读取前检查取消状态。上行解码器只保留尚未完成的读取：取消时拒绝这些读取，正常关闭则以 `done` 唤醒全部读取，包括重叠的读取。Client 泵在每次读取后移除停止 Promise，停止后不再发起读取。源迭代器的释放仍由原有所有者负责；Host 仍先关闭上行，再返回方法迭代器。
+
 ### 顺序保证
 
 同一条 socket 上的帧全序到达：一条逻辑流的 `open`、`item`…、`end` 按发送顺序到达 Host；上行项进入 `uplink()` 的顺序就是 `send` 的顺序。上行与下行之间没有跨方向顺序保证；需要请求应答配对的协议自带序号。
@@ -538,6 +544,10 @@ for await (const reply of stream) replies.push(reply)   // ['> a', '> b']
 
 ## 考虑过的替代方案
 
+**三处读取循环共用一个取消控制器。** 每处循环已有自己的所有者和不同的关闭行为。单次读取的 Promise 能限制保留状态，不需要把这些职责迁入新抽象。
+
+**要求源迭代器自行取消 `next()`。** Gateway 接受任意 iterable，其待处理读取不一定响应取消或 `return()`。Gateway 仍需独立于源清理来唤醒自己的等待者。
+
 **独立的 `mode: 'duplex'` 加一个与 `signal` 同级的保留参数 `uplink: AsyncIterable<In>`。** 每一处都是特殊逻辑：analyzer 要认第二个保留参数及其位置规则，生成签名多一个参数，加载期要校验 `duplex ⟺ uplink`，mux 不知道方法模式所以网关要替 `'stream'` 方法拒绝上行帧。双工是传输层流的固有性质，不是方法的一种模式：载体本来就有 `streamId`、`open`、`cancel`，上行只是再加两种帧，任何流都能收，不用就不发。
 
 **用装饰器选项声明上行类型，`@Remote({ mode: 'stream', uplink: 'JobInputFrame' })`。** 字符串引用类型名不受编译器约束，重命名即断。返回类型别名把两个类型写在同一个编译器可见的位置。
@@ -572,6 +582,8 @@ for await (const reply of stream) replies.push(reply)   // ['> a', '> b']
 
 ## 测试
 
+`gateway-stream.host.spec.ts` 和 `gateway.client.spec.ts` 中的保留对象用例在流保持打开时进行全量 GC，再统计存活的迭代结果。生命周期用例覆盖正常关闭唤醒多个待处理上行读取、两次读取之间的取消、源同步取消后抛错，以及 Client 发送期间流结束。
+
 - `packages/api/gateway/tests/echo-stream.host.spec.ts`：同一个 echo 方法分别穿过 WebSocket 载体与进程内载体。
 - 网关 Host：帧解析的精确键校验；`UplinkInbox` 的 opening 缓冲、`end` 后 `item`、超限、未知 `streamId` 忽略、socket 关闭时的收尾；`UplinkDecoder` 有无 codec 两种交付与解码失败中止；`uplink()` 只能取一次；`extend({ invocation })` 视图与 `peer` 的来源；`cancellableStream` 先关 uplink 再 `return()`。
 - 网关 Client：句柄的 `send`（JSON 校验、open 前入队、终止后抛错）、`end` 幂等、`dispose` 驱动 `cancel` 帧并结束后续读取；泵的三种退出；终止帧到达即停泵。

packages/api/gateway/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/api/gateway/README.md
-README.md: 1a746a83eb966fba6ae40b61061d42bf478e1c84
-README.zh.md: 7b37548f6ad894b761457df80278eeed7aa6ca7b
+README.md: e0c986e290b1744017fca99e3a2553a68d754ef6
+README.zh.md: 8be11a7c5aea1e299c4ca45cde4b1c9dfa08ec4b

packages/api/gateway/README.md

@@ -38,6 +38,8 @@ When the launcher supplies `ctx.appReady`, Gateway registers the WebSocket upgra
 
 Every stream also carries a Client-to-Host uplink on the same logical stream. A method declares the uplink item type as the second type argument of its return type, `RemoteStream<Out, In>` from [`dsh-typert-protocol`](../../typert/protocol/README.md); `In` defaults to `never`, and a method that returns `Iterable`, `AsyncIterable`, or `RemoteStream<Out>` declares that it reads none. The running method reads its call as `this.ctx.invocation`: the Gateway derives the receiver from a Context carrying that `RemoteInvocation` — `request`, `service`, `peer`, `signal`, and `uplink()` — so nothing enters the parameter list, and `ctx.invocation` is `undefined` on a Context no Remote call derived. `uplink()` is available once per call. With a descriptor codec it decodes every item before delivery; without one — an SRC method, or a method whose `In` is `never` — it delivers JSON-safe `unknown` values. The Client sends each item as an `item` frame on the open logical stream, a top-level `undefined` item as an `item` frame without `value`, and its half-close as an `end` frame; the method observes the half-close as the end of the `uplink()` iteration and cancellation through `signal`. A rejected item fails the whole stream with `gateway/input-invalid` (field `uplink`). Uplink items are validated one by one at the Host because they arrive from the browser; downlink items are values the Host method produced and pass through unchecked. When the method finishes its downlink the Gateway returns the uplink iterator and drops unread items; items sent to a method that never takes its uplink wait in the inbox until then. Each logical stream buffers at most `streamInboxBytes` (262144 by default) of uplink frame bytes; overflow fails the stream with `gateway/uplink-overflow`, and an item after `end` fails it with `gateway/protocol`, in both cases without closing the socket. `item`, `end`, and `cancel` frames for a stream id the Host has already finished are dropped; only a duplicate `open` closes the socket. Gateway-owned streams such as `$events` return their uplink as soon as they open, so their `item` frames are dropped instead of buffered. A unary method may also call `uplink()`; its items are readable only while the method runs. Every stream speaks for the operator Peer: a WebSocket's streams for the Peer Connection admitted at upgrade, whose scope disposal closes the socket, and an in-process carrier's for the operator directly.
 
+Stream cancellation and stopping keep read-local state, not a history of delivered items, on the Host downlink, Host uplink, and Client uplink pump. Completing the downlink also wakes pending uplink reads.
+
 Host composition can register one application event source through `registerRemoteEvents()`. Gateway reserves the internal `$events` logical endpoint for that source, accepts only empty `args`, and aborts streams opened by the registration when the source is withdrawn. API Remotes owns the event selection, argument validation, per-Client queues, and the Host home sent in the opening `{ type: 'ready', clientId, host: { home } }` frame. Its source factory attaches incremental listeners synchronously, so the Client publishes the generation and starts baseline reads only after incremental delivery is ready.
 
 <a id="client-service-clientremote-ctx-key-remote"></a>

packages/api/gateway/README.zh.md

@@ -38,6 +38,8 @@ Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandle
 
 每条流还在同一条逻辑流上携带从 Client 到 Host 的上行。方法把上行项类型声明为返回类型的第二个类型参数——来自 [`dsh-typert-protocol`](../../typert/protocol/README.zh.md) 的 `RemoteStream<Out, In>`；`In` 缺省为 `never`，返回 `Iterable`、`AsyncIterable` 或 `RemoteStream<Out>` 的方法即声明自己不读上行。运行中的方法通过 `this.ctx.invocation` 读取本次调用：Gateway 从携带该 `RemoteInvocation`（`request`、`service`、`peer`、`signal` 与 `uplink()`）的 Context 派生接收者，因此没有任何东西进入参数列表，而非 Remote 调用派生的 Context 上 `ctx.invocation` 为 `undefined`。`uplink()` 每次调用只能取一次。描述符带 codec 时它在交付前逐项解码；不带时——SRC 方法，或 `In` 为 `never` 的方法——它交付经 JSON 安全校验的 `unknown` 值。Client 在已打开的逻辑流上把每一项作为 `item` 帧发送（顶层 `undefined` 项是不带 `value` 的 `item` 帧），以 `end` 帧表示半关闭；方法通过 `uplink()` 迭代结束感知半关闭，通过 `signal` 感知取消。被拒绝的项使整条流以 `gateway/input-invalid`（字段 `uplink`）失败。上行项在 Host 逐项校验，因为它们来自浏览器；下行项是 Host 方法产出的值，不经校验原样透传。方法结束下行时 Gateway 调用 uplink 迭代器的 return 并丢弃未读的项；发给从未取用 uplink 的方法的项在 inbox 里等到那时为止。每条逻辑流最多缓冲 `streamInboxBytes`（默认 262144）字节的上行帧；超限使流以 `gateway/uplink-overflow` 失败，`end` 之后的项使流以 `gateway/protocol` 失败，两种情况都不关闭 socket。发给 Host 已结束的流 id 的 `item`、`end` 与 `cancel` 帧被丢弃；只有重复的 `open` 会关闭 socket。`$events` 这类 Gateway 自有的流在打开时立即 return 其 uplink，因此它们的 `item` 帧被丢弃而不缓冲。一元方法也可以调用 `uplink()`；其项只在方法运行期间可读。每条流都代表操作者 Peer：WebSocket 上的流代表 Connection 在升级时接纳的 Peer，该 scope 释放时 socket 关闭；进程内载体的流直接代表操作者。
 
+Host 下行、Host 上行和 Client 上行泵的取消与停止状态归属单次读取，不保留已交付项的历史。下行结束也会唤醒尚未完成的上行读取。
+
 Host 组合可通过 `registerRemoteEvents()` 注册唯一的应用事件 source。Gateway 为它保留内部 `$events` logical endpoint，只接受空 `args`，并在 source 撤回时中止该注册打开的流。事件名单、参数校验、每个 Client 的队列及 opening `{ type: 'ready', clientId, host: { home } }` frame 中的 Host home 由 API Remotes 拥有。source factory 在返回 iterable 前同步挂好增量 listener，因此 Client 只在增量投递就绪后发布 generation 并开始 baseline 读取。
 
 <a id="client-service-clientremote-ctx-key-remote"></a>

packages/api/gateway/src/client/stream-client.ts

@@ -158,7 +158,7 @@ export class RemoteStreamMuxClient {
     signal: AbortSignal,
     inbox: StreamInbox,
   ): UplinkPump {
-    const stopped = Promise.withResolvers<IteratorReturnResult<undefined>>()
+    let stopped: PromiseWithResolvers<IteratorReturnResult<undefined>> | undefined
     const interruption: IteratorReturnResult<undefined> = { value: undefined, done: true }
     const uplinkIterator = uplink[Symbol.asyncIterator]()
     const state = { stopping: false, exhausted: false, released: false }
@@ -171,7 +171,12 @@ export class RemoteStreamMuxClient {
     const done = (async (): Promise<void> => {
       try {
         while (true) {
+          const stoppingBeforeRead = state.stopping
+          if (stoppingBeforeRead) return
+          // A stop promise belongs to one read, not the whole uplink history.
+          stopped = Promise.withResolvers<IteratorReturnResult<undefined>>()
           const next = await Promise.race([uplinkIterator.next(), stopped.promise])
+          stopped = undefined
           if (state.stopping || next === interruption || signal.aborted || this.socket !== socket) return
           if (next.done === true) {
             state.exhausted = true
@@ -183,6 +188,7 @@ export class RemoteStreamMuxClient {
       } catch (error) {
         inbox.fail(error)
       } finally {
+        stopped = undefined
         release()
       }
     })()
@@ -191,7 +197,7 @@ export class RemoteStreamMuxClient {
       stop: () => {
         if (state.stopping) return
         state.stopping = true
-        stopped.resolve(interruption)
+        stopped?.resolve(interruption)
         release()
       },
     }

packages/api/gateway/src/index.ts

@@ -1164,19 +1164,24 @@ async function *cancellableStream(
     ? Reflect.apply(asyncFactory, source, []) as AsyncIterator<unknown>
     : Reflect.apply(syncFactory as (...args: never[]) => Iterator<unknown>, source, [])
   let rejectAbort: ((error: unknown) => void) | undefined
-  const aborted = new Promise<never>((_resolve, reject) => { rejectAbort = reject })
   const onAbort = (): void => {
     rejectAbort?.(streamAbortFailure(endpoint, signal.reason))
   }
   signal.addEventListener('abort', onAbort, { once: true })
   try {
-    if (signal.aborted) throw streamAbortFailure(endpoint, signal.reason)
     while (true) {
+      if (signal.aborted) throw streamAbortFailure(endpoint, signal.reason)
+      // Reusing a pending cancellation promise retains every completed race.
+      const aborted = new Promise<never>((_resolve, reject) => { rejectAbort = reject })
+      // next() can abort and throw synchronously before the race subscribes.
+      void aborted.catch(() => undefined)
       const next = await Promise.race([Promise.resolve(iterator.next()), aborted])
+      rejectAbort = undefined
       if (next.done === true) return
       yield next.value
     }
   } finally {
+    rejectAbort = undefined
     signal.removeEventListener('abort', onAbort)
     // The uplink closes first so a method blocked on `uplink.next()` unwinds
     // before its iterator is asked to return.
@@ -1209,9 +1214,10 @@ function remoteCancelled(endpoint: string, cause: unknown): RemoteError<'gateway
  */
 class UplinkDecoder implements AsyncIterable<unknown>, AsyncIterator<unknown> {
   private readonly source: AsyncIterator<unknown>
-  private readonly interrupted = Promise.withResolvers<IteratorResult<unknown>>()
+  private readonly interrupted = new Set<PromiseWithResolvers<IteratorResult<unknown>>>()
   private readonly onAbort = (): void => {
-    this.interrupted.reject(streamAbortFailure(this.endpoint, this.signal.reason))
+    const failure = streamAbortFailure(this.endpoint, this.signal.reason)
+    for (const read of this.interrupted) read.reject(failure)
   }
   private closed = false
 
@@ -1223,9 +1229,6 @@ class UplinkDecoder implements AsyncIterable<unknown>, AsyncIterator<unknown> {
     private readonly abort: (reason: unknown) => void,
   ) {
     this.source = uplink[Symbol.asyncIterator]()
-    // The rejection settles the race inside next(); an abort with no read in
-    // flight must not surface as an unhandled rejection.
-    void this.interrupted.promise.catch(() => undefined)
     signal.addEventListener('abort', this.onAbort, { once: true })
   }
 
@@ -1237,7 +1240,17 @@ class UplinkDecoder implements AsyncIterable<unknown>, AsyncIterator<unknown> {
     // A cancelled stream reports the cancellation on every read, closed or not.
     if (this.signal.aborted) throw streamAbortFailure(this.endpoint, this.signal.reason)
     if (this.closed) return UPLINK_DONE
-    const next = await Promise.race([this.source.next(), this.interrupted.promise])
+    // Only pending reads belong to the decoder; finish() must wake all of them.
+    const interrupted = Promise.withResolvers<IteratorResult<unknown>>()
+    this.interrupted.add(interrupted)
+    // The source can abort and throw synchronously before the race subscribes.
+    void interrupted.promise.catch(() => undefined)
+    let next: IteratorResult<unknown>
+    try {
+      next = await Promise.race([this.source.next(), interrupted.promise])
+    } finally {
+      this.interrupted.delete(interrupted)
+    }
     if (next.done === true) {
       this.finish()
       return UPLINK_DONE
@@ -1270,7 +1283,7 @@ class UplinkDecoder implements AsyncIterable<unknown>, AsyncIterator<unknown> {
   private finish(): void {
     this.closed = true
     this.signal.removeEventListener('abort', this.onAbort)
-    this.interrupted.resolve(UPLINK_DONE)
+    for (const read of this.interrupted) read.resolve(UPLINK_DONE)
   }
 }
 

packages/api/gateway/tests/gateway-stream.host.spec.ts

@@ -1,5 +1,6 @@
 import { randomUUID } from 'node:crypto'
 import { once } from 'node:events'
+import { queryObjects } from 'node:v8'
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import WebSocket, { type RawData } from 'ws'
 import { Context, symbols } from '@deepseek-ai/cordis'
@@ -56,6 +57,7 @@ class FeedService extends TypertRemoteService {
   readonly peeked: unknown[] = []
   readonly peers: PeerScope[] = []
   leftover: AsyncIterator<string> | undefined
+  source: Iterable<string> = []
   returns = 0
 
   constructor(ctx: Context) {
@@ -82,6 +84,11 @@ class FeedService extends TypertRemoteService {
     yield `${label}:two`
   }
 
+  @Remote({ mode: 'stream' })
+  items(): Iterable<string> {
+    return this.source
+  }
+
   @Remote({ mode: 'stream' })
   *invalid(): Iterable<string> {
     yield 42 as unknown as string
@@ -541,6 +548,110 @@ describe('Typert Remote streams', () => {
     await expect(iterator.return?.()).resolves.toEqual({ done: true, value: undefined })
   })
 
+  it('finishes every pending uplink read when the downlink returns', async () => {
+    const { ctx, service } = await setup(false)
+    const reads = [
+      Promise.withResolvers<IteratorResult<string>>(),
+      Promise.withResolvers<IteratorResult<string>>(),
+    ]
+    let index = 0
+    const returned = vi.fn(async (): Promise<IteratorResult<string>> => ({ done: true, value: undefined }))
+    const source = await ctx.typertGateway.stream({
+      namespace: 'feed', method: 'hold', args: {},
+      uplink: { [Symbol.asyncIterator]: () => ({ next: () => reads[index++]!.promise, return: returned }) },
+    })
+    const iterator = source[Symbol.asyncIterator]()
+    const completed: IteratorResult<string>[] = []
+    const pending: Promise<void>[] = []
+    try {
+      await expect(iterator.next()).resolves.toEqual({ done: false, value: 'held' })
+      const uplink = service.leftover!
+      pending.push(...reads.map(() => uplink.next().then((result) => { completed.push(result) })))
+      await iterator.return?.()
+      expect(completed).toEqual([
+        { done: true, value: undefined },
+        { done: true, value: undefined },
+      ])
+      expect(returned).toHaveBeenCalledOnce()
+    } finally {
+      for (const read of reads) read.resolve({ done: true, value: undefined })
+      await Promise.all(pending)
+      await iterator.return?.()
+    }
+  })
+
+  it.each(['downlink', 'uplink'] as const)('releases consumed %s read results while the stream stays open', async (direction) => {
+    class ReadResult implements IteratorYieldResult<string> {
+      readonly done = false
+      readonly value = 'item'
+    }
+    const { ctx, service } = await setup(false)
+    service.source = { [Symbol.iterator]: () => ({ next: () => new ReadResult() }) }
+    const source = await ctx.typertGateway.stream(direction === 'downlink'
+      ? { namespace: 'feed', method: 'items', args: {} }
+      : {
+        namespace: 'feed', method: 'echo', args: { prefix: '' },
+        uplink: { [Symbol.asyncIterator]: () => ({ next: async () => new ReadResult() }) },
+      })
+    const iterator = source[Symbol.asyncIterator]()
+    try {
+      for (const count of [64, 256]) {
+        for (let index = 0; index < count; index++) {
+          await expect(iterator.next()).resolves.toEqual({ done: false, value: 'item' })
+        }
+        // queryObjects collects garbage; the suspended read can retain its last result, not the history.
+        expect(queryObjects(ReadResult, { format: 'count' })).toBeLessThanOrEqual(2)
+      }
+    } finally {
+      await iterator.return?.()
+    }
+  })
+
+  it('does not read another downlink item after cancellation between reads', async () => {
+    const { ctx, service } = await setup(false)
+    const abort = new AbortController()
+    const next = vi.fn((): IteratorResult<string> => ({ done: false, value: 'item' }))
+    service.source = { [Symbol.iterator]: () => ({ next }) }
+    const source = await ctx.typertGateway.stream({
+      namespace: 'feed', method: 'items', args: {}, signal: abort.signal,
+    })
+    const iterator = source[Symbol.asyncIterator]()
+    try {
+      await expect(iterator.next()).resolves.toEqual({ done: false, value: 'item' })
+      abort.abort(new Error('caller cancelled between reads'))
+      await expect(iterator.next()).rejects.toMatchObject({ code: 'gateway/cancelled' })
+      expect(next).toHaveBeenCalledOnce()
+    } finally {
+      await iterator.return?.()
+    }
+  })
+
+  it.each(['downlink', 'uplink'] as const)('handles a synchronous %s read that aborts and throws', async (direction) => {
+    const { ctx, service } = await setup(false)
+    const abort = new AbortController()
+    const failure = new Error('source aborted and threw')
+    const next = (): never => {
+      abort.abort(failure)
+      throw failure
+    }
+    service.source = { [Symbol.iterator]: () => ({ next }) }
+    const source = await ctx.typertGateway.stream(direction === 'downlink'
+      ? { namespace: 'feed', method: 'items', args: {}, signal: abort.signal }
+      : {
+        namespace: 'feed', method: 'hold', args: {}, signal: abort.signal,
+        uplink: { [Symbol.asyncIterator]: () => ({ next }) },
+      })
+    const iterator = source[Symbol.asyncIterator]()
+    try {
+      if (direction === 'uplink') {
+        await expect(iterator.next()).resolves.toEqual({ done: false, value: 'held' })
+      }
+      await expect((direction === 'downlink' ? iterator : service.leftover!).next()).rejects.toBe(failure)
+    } finally {
+      await iterator.return?.()
+    }
+  })
+
   it('reports done on every read after the uplink ended', async () => {
     const { ctx } = await setup(false)
     await expect(collect(await ctx.typertGateway.stream({
@@ -1461,6 +1572,7 @@ function descriptors(): InvocationDescriptor[] {
     stream('context', [label], z.string()),
     { ...stream('follow', [label], z.string()), cancellation: { parameter: 'signal' } },
     stream('sync', [label], z.string()),
+    stream('items', [], z.string()),
     stream('invalid', [], z.string()),
     stream('nonJson', [], z.unknown()),
     stream('missing', [], z.string()),

packages/api/gateway/tests/gateway.client.spec.ts

@@ -1,3 +1,4 @@
+import { queryObjects } from 'node:v8'
 import { RemoteError, typertOwnedValue } from '@deepseek-ai/dsh-typert-protocol'
 import { Context, Service } from '@deepseek-ai/cordis'
 import type { Fiber } from '@deepseek-ai/cordis'
@@ -2804,6 +2805,81 @@ describe('Client Typert API', () => {
 })
 
 describe('Remote stream client carrier lifecycle', () => {
+  it('does not pull another uplink item when sending ends the stream', async () => {
+    await withFakeWebSocket('https://harness.example', async () => {
+      const client = new RemoteStreamMuxClient()
+      client.start()
+      const socket = FakeWebSocket.sockets[0]!
+      const send = socket.send.bind(socket)
+      const sent = vi.spyOn(socket, 'send').mockImplementation((text) => {
+        send(text)
+        const frame = parseRemoteStreamClientMessage(text)
+        if (frame.type === 'item') socket.receive({ type: 'end', streamId: frame.streamId })
+      })
+      const next = vi.fn(async (): Promise<IteratorResult<string>> => ({ done: false, value: 'item' }))
+      const returned = vi.fn(async (): Promise<IteratorResult<string>> => ({ done: true, value: undefined }))
+      try {
+        const iterator = client.open('probe/attach', {}, new AbortController().signal, {
+          [Symbol.asyncIterator]: () => ({ next, return: returned }),
+        })[Symbol.asyncIterator]()
+        await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined })
+        expect(next).toHaveBeenCalledOnce()
+        expect(returned).toHaveBeenCalledOnce()
+      } finally {
+        await client.close()
+        sent.mockRestore()
+      }
+    })
+  })
+
+  it('releases sent uplink read results while the stream stays open', async () => {
+    await withFakeWebSocket('https://harness.example', async () => {
+      class ReadResult implements IteratorYieldResult<string> {
+        readonly done = false
+        readonly value = 'item'
+      }
+      const blocked = Promise.withResolvers<undefined>()
+      const lastRead = Promise.withResolvers<IteratorResult<string>>()
+      let reads = 0
+      const uplink: AsyncIterable<string> = {
+        [Symbol.asyncIterator]: () => ({
+          next: () => {
+            if (reads++ < 256) return Promise.resolve(new ReadResult())
+            blocked.resolve(undefined)
+            return lastRead.promise
+          },
+          return: async () => {
+            lastRead.resolve({ done: true, value: undefined })
+            return { done: true, value: undefined }
+          },
+        }),
+      }
+      const client = new RemoteStreamMuxClient()
+      client.start()
+      const iterator = client.open('probe/attach', {}, new AbortController().signal, uplink)[Symbol.asyncIterator]()
+      const first = iterator.next()
+      // Closing the carrier during failed assertions can reject this pending read.
+      void first.catch(() => undefined)
+      try {
+        await blocked.promise
+        const socket = FakeWebSocket.sockets[0]!
+        expect(socket.sent).toHaveLength(257)
+        const opened = parseRemoteStreamClientMessage(socket.sent[0]!)
+        if (opened.type !== 'open') throw new Error('fixture expected an open frame')
+        socket.receive({ type: 'item', streamId: opened.streamId, value: 'ready' })
+        await expect(first).resolves.toEqual({ done: false, value: 'ready' })
+        // A full GC runs while the uplink pump is still waiting for its next item.
+        expect(queryObjects(ReadResult, { format: 'count' })).toBeLessThanOrEqual(2)
+        socket.receive({ type: 'end', streamId: opened.streamId })
+        await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined })
+      } finally {
+        await client.close()
+        await first.catch(() => undefined)
+        await iterator.return?.(undefined)
+      }
+    })
+  })
+
   it('connects to the shell-owned Host while the document uses a local asset origin', async () => {
     await withFakeWebSocket('dsh-app://app', async () => {
       vi.stubGlobal('__DSH_TRANSPORT__', { streamBaseUrl: 'http://127.0.0.1:43210' })