2026-09-19implementedarchitecture

Remote duplex stream: one kind of stream, an uplink channel, and an invocation context

Remote 双工流：一种流、上行通道与调用上下文

### Current state `dsh-api-gateway` multiplexes Typert Remote streams over one WebSocket at `/api/remote.mux`. A Host method declares both directions with `@Remote({ mode: 'stream' })` and `RemoteStream<Out, In>`; the generated Client method returns `RemoteStreamHandle<Out, In>`. Existing `Iterable` and `AsyncIterable` return types declare downlink-only methods. The optional final `signal: AbortSignal` is supplied by

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

English

Problem

Current state

dsh-api-gateway multiplexes Typert Remote streams over one WebSocket at /api/remote.mux. A Host method declares both directions with @Remote({ mode: 'stream' }) and RemoteStream<Out, In>; the generated Client method returns RemoteStreamHandle<Out, In>. Existing Iterable and AsyncIterable return types declare downlink-only methods. The optional final signal: AbortSignal is supplied by the Gateway and does not enter the wire arguments.

Both directions share one streamId:

Direction	Frame	Definition
Client to Host	`{ type: 'open', streamId, endpoint, payload }`	`stream-protocol.ts`
Client to Host	`{ type: 'item', streamId, value? }`	`stream-protocol.ts`
Client to Host	`{ type: 'end', streamId }`	`stream-protocol.ts`
Client to Host	`{ type: 'cancel', streamId }`	`stream-protocol.ts`
Host to Client	`{ type: 'item', streamId, value? }`	`stream-protocol.ts`
Host to Client	`{ type: 'error', streamId, error: { code, message, details } }`	`stream-protocol.ts`
Host to Client	`{ type: 'end', streamId }`	`stream-protocol.ts`

The Client handle's send() supplies uplink items, end() half-closes the uplink, and dispose() cancels the logical stream. The Host reads decoded items through this.ctx.invocation.uplink<In>(); the invocation also carries the admitted Peer and cancellation signal.

The transport owns validation, cancellation, and its bounded inbox. Consumer protocols own input authority, acknowledgements, and recovery after disconnection.

Uplink consumers

Feature	Uplink path	Location	Cost
Web terminal input and resize	`terminal.attach` carries `TerminalInput` uplink items and `TerminalFrame` downlink items; the Client pipelines commands and releases its raw-input byte budget on `ack`	`packages/api/terminal-controller/src/client/model.ts`; Host `src/terminal.ts`	The current attachment owns input; `inputFull` when the Client budget is exhausted; input is never replayed after reconnection
File upload	A `Blob` or `ReadableStream` goes to the native `POST /api/session/uploadFileBinary` outside the Remote layer, with `duplex: 'half'`	`packages/client/file-upload/src/client/runtime.ts`; route `src/index.ts`	Bypasses Typert descriptors; no type projection
Answers to approvals and questions	The Host sends waterfall frames down the `$events` stream, the Client answers through a separate unary RPC `$events/result`, correlated by `clientId` plus `eventId`	`packages/api/gateway/src/client/remote-events.ts`; Host `index.ts`	A second correlation key; `index.ts` records the race this creates

These consumers need Client-to-Host data while a related operation remains active. Web terminals use the duplex stream; file upload and event answers retain their own transports.

Consumer responsibilities

Input authority. A terminal retains its controlling attachment as a Host-local follower. Snapshot and state frames report each attachment's writable state; older attachments keep receiving output after control transfers.
Ordering and acknowledgement. Terminal writes and resizes share a consecutive command sequence and the terminal's operation queue. An ack reports provider acceptance or a control refusal; it never authorizes replay after disconnection.
Independent lifetimes. Cancelling attach stops its input and output delivery, drops unstarted commands, and joins an in-flight input operation. The PTY lifetime is owned separately by retention and explicit close.
Coordinated byte limits. The Gateway exposes its complete-frame inbox capacity and this stream's JSON envelope cost through invocation.uplinkWindow. The terminal publishes that window in its snapshot and uses provider acknowledgements to release in-flight capacity. Writes and resizes wait in a bounded Client queue when the window is full. maxInputBytes limits raw UTF-8 input; maxBufferedBytes separately limits the complete encoded input retained by the Client and output retained by each follower.
Remaining separate carriers. Approval answers still correlate $events with the $events/result RPC through clientId and eventId. Moving this protocol onto the duplex stream remains deferred.

Relation to background jobs

The process-shaped duplex stream face of background jobs (archived as a draft PR on branch worktree/job-stdio-stream-pair-archive) models a job as a pair of mirrored stream faces; for the browser to attach to it, one Remote stream must carry both a downlink and an uplink. It is one motivation for this Note but not in its scope: whether jobs need a dedicated read-write stream is an architecture question for the jobs line itself. This Note covers only the transport layer.

Goals

The transport layer has one kind of stream, duplex by nature: downlink item / end / error, uplink item / end, one streamId, one generation.
The uplink type and the downlink type are written in the same place: the method's return type.
A Host method takes its uplink from the invocation context; the invocation context also carries the caller (the Peer).
The Client receives a handle that can read, write, and close, not a bare AsyncIterable.
Upper-layer protocols (snapshot, journal, any business-defined stream) stay out of the transport layer.

Decision

In one sentence

Every Remote stream accepts a Client uplink. The method's return type RemoteStream<Out, In = never> declares the downlink item type and the uplink item type together; the generator derives two codecs from it. A Host method takes the uplink iterator through this.ctx.invocation.uplink<In>() and learns the caller through this.ctx.invocation.peer. The generated Client method returns a RemoteStreamHandle<Out, In>: for await reads the downlink, send / end write the uplink, dispose closes. Reopening after disconnection, cursor resumption, and baseline validation all belong to upper-layer protocols.

                 Client                                             Host
   ─────────────────────────────────              ───────────────────────────────────────────
   const s = remote.job.attach(req, signal)
        │  open { streamId, endpoint, payload } ───────▶  opener(endpoint, payload, uplink, peer, control, window)
        │                                                     │ prepareInvocation
        │  s.send(v1) → item { streamId, v1 }    ───────▶  inbox.push(v1)   │ invocation = { request, service, peer, signal, uplink() }
        │  s.send(v2) → item { streamId, v2 }    ───────▶  inbox.push(v2)   │ receiverContext.extend({ invocation }).get(service)
        │  s.end()    → end  { streamId }        ───────▶  inbox.end()      │ method(...args, signal)
        │                                        ◀───────  item { streamId, o1 }   │   for await (v of this.ctx.invocation.uplink()) …
        │  for await (o of s) …                  ◀───────  item { streamId, o2 }   │   yield o …
        │  s.dispose() → cancel { streamId }     ───────▶  control.abort()         │ signal aborts; uplink.return()
        │                                        ◀───────  end | error

Terms

Term	Meaning
Logical stream	One call identified by one `streamId`, spanning `open` to the terminal frame
Generation	One physical lifetime of a logical stream. The transport layer knows exactly one generation; disconnection is failure, and reopening belongs to the upper layer
Downlink	The Host-to-Client `item` sequence, terminated by `end` or `error`
Uplink	The Client-to-Host `item` sequence, terminated by `end` (half-close) or ended together with `cancel`
Half-close	One direction has ended while the other remains open
Item	One business value in one direction; every item is validated by the codec independently
inbox	The bounded uplink queue the Host keeps per logical stream
Peer	The party one connection speaks for, one `PeerScope` carrying its principal; every browser connection shares `connection.browserPeer`, and in-process code creates its own through `connection.createPeer()` (Peer principals)
Invocation context	`RemoteInvocation`: the request, receiving service, Peer, cancellation signal, and uplink entry of this call

Type signatures

// @deepseek-ai/dsh-typert-protocol
/**
 * One Remote stream. Host face: the method returns it, and at runtime it is AsyncIterable<Out>.
 * On the Client face the generated method returns RemoteStreamHandle<Out, In>; each name has exactly one meaning.
 * In is the item type the Client may send uplink; the default never means the method reads no uplink.
 */
export type RemoteStream<Out, In = never> = AsyncIterable<Out> & { readonly [STREAM_UPLINK]?: In }

// Host
@Remote({ mode: 'stream' })
async *attach(request: JobAttachRequest, signal: AbortSignal): RemoteStream<JobFollowFrame, JobInputFrame> {
  const uplink = this.ctx.invocation.uplink<JobInputFrame>()
  void (async () => { for await (const frame of uplink) … })()
  yield …
}

// Generated Client signature: same parameters, returns the handle
attach(request: JobAttachRequest, signal?: AbortSignal): RemoteStreamHandle<JobFollowFrame, JobInputFrame>

// Client usage
const stream = remote.job.attach(req, signal)
for await (const frame of stream) …
stream.send(frame)     // typed as JobInputFrame
stream.end()
stream.dispose()

RemoteStream is an alias rather than an interface, deliberately: if the transport layer later takes a Node stream form, only the alias definition and the adapters at both ends change; method signatures, descriptors, and upper-layer protocols stay untouched.

Host face

/** The context of this Remote call. A method reads it through this.ctx.invocation. */
export interface RemoteInvocation {
  readonly request: {
    readonly namespace: string
    readonly method: string
    readonly args: Readonly<Record<string, unknown>>
  }
  /** Cordis service key of the receiving service. */
  readonly service: string
  /** The Peer that initiated the call; every carrier names it explicitly. */
  readonly peer: PeerScope
  /** Carrier cancellation: Client cancel, socket closure, uplink failure. */
  readonly signal: AbortSignal
  readonly uplinkWindow?: RemoteUplinkWindow | undefined
  /**
   * The uplink items of this call. Can be taken once; the second call throws. With an uplink codec
   * in the descriptor each item is decoded to In; without one, items are delivered as unknown with
   * only a JSON-safety check.
   * The iteration ends after the Client's end; when the method ends the downlink, the Gateway calls
   * its return() and drops unconsumed items.
   * The generic In is only the caller's type assertion; runtime decodes per the descriptor and does not cross-check.
   */
  uplink<In = unknown>(): AsyncIterable<In>
}

declare module '@deepseek-ai/cordis' {
  interface Context {
    /** The Remote call this Context was derived for; undefined on a Context not derived from a Remote call. */
    readonly invocation: RemoteInvocation | undefined
  }
}

Rules:

Any mode: 'stream' method may call this.ctx.invocation.uplink(); a unary method may too, but its stream ends when the method returns, so uplink items are readable only while the method runs.
While uplink() has not been called, uplink items accumulate in the inbox until the limit is exceeded (see Backpressure) or the stream ends and they are dropped. This is deliberate: a method that does not read the uplink need not know the uplink exists.
When In defaults to never, the descriptor has no uplink codec; the Client handle's send has type never, so nothing can be sent at compile time. If frames still arrive at runtime (from a hand-built client), they are delivered as unknown.
this.ctx.invocation is undefined on a Context not derived from a Remote call; a method that reads it handles the optional as TypeScript requires, or asserts it on the method's first line. The accessor for that property is registered once when the TypertRemoteService base class is constructed, so any composition that installs a Remote service can read it.

How the invocation context reaches the method

The mechanism relies on two existing Cordis behaviors:

ctx.extend(meta) is Object.create(parent) plus own properties (vendor/cordis/src/context.ts:99-107): no registration, no fiber, nothing to release, so deriving one per call costs nothing.
ctx.get(service) returns the createTraceable(ctx, service) proxy (vendor/cordis/src/reflect.ts:233-234, utils.ts:117-125). Reading the ctx property returns the accessing Context directly (utils.ts:175); reading a method wraps it in createShadowMethod, whose apply replaces this with a shadow receiver whose ctx is ctx.extend({ [symbols.shadow]: origin }) (utils.ts:149-163).
The Gateway therefore writes receiverContext.extend({ invocation }).get(descriptor.service) to obtain a per-call view and calls Reflect.apply(method, callReceiver, args); inside the method body, this.ctx.invocation reads this call through the prototype chain. The deferred execution of an async generator body is unaffected, because this was bound to the shadow object at call time.

PeerScope

/** The opaque identity of one Peer. */
export type PeerId = Branded<'PeerId'>

/** One Peer's session on this Host: opened and released by its creator; ctx owns connection-lifetime registrations. */
export interface PeerScope {
  readonly id: PeerId
  readonly ctx: Context
  /** Caller identity fixed at creation. */
  readonly principal: Principal
  /** Aborted once the Peer is disposed. */
  readonly signal: AbortSignal
  /** Participation in forwarded interactions; independent of RBAC. */
  readonly interactions: 'answer' | 'none'
  dispose(): Promise<void>
}

@deepseek-ai/dsh-client-connection implements the one Peer class, Peer(owner, { principal, interactions? }), with createScope(owner, peer), the same mechanism an Agent uses to create its own scope: the Peer object is the ScopeKey, peer.ctx carries connection-lifetime effects, peer.signal aborts when the scope ends, and dispose() quiets the fiber. Connection creates connection.browserPeer with the owner principal when the service applies; Peer principals and method-level Remote access control owns principals, createPeer(), and access checks.

Member	Semantics
`connection.browserPeer: PeerScope`	The Peer every authenticated browser connection shares; owner principal; lives and dies with the Connection
`connection.admit(request): PeerAdmission`	Checks Host/Origin and signed browser or API Cookies; returns `{ rejection: 401 \| 403 }` or the credential's live Peer; connect publishes the API Peer and its lifecycle socket releases it

The Gateway's handleUpgrade(req, socket, head, peer) binds an opener with a fixed Peer to that socket and registers socket closure in peer.ctx.effect; when the scope is already inactive, it closes the socket directly with 1001 instead. InvokeRemoteRequest.peer and the peer parameter of RemoteStreamOpener and wireStream.open are required, so every in-process caller names its Peer, and the Gateway creates none. Looking up a Peer by id and Peer open and close events wait until the first consumer that needs them appears.

The Peer carries its principal; what that principal may call is decided by the roles, bindings, and policies that Peer principals and method-level Remote access control defines.

Client face

/** The handle a generated method returns on the Client face. A different name from the Host's RemoteStream, exported from the same entry point. */
export interface RemoteStreamHandle<Out, In> extends AsyncIterable<Out> {
  /** Send one uplink item. Throws on a value that is not lossless JSON, after termination, or after end. */
  send(item: In): void
  /** Half-close the uplink: sends the end frame. Idempotent. */
  end(): void
  /** Cancel the whole logical stream: sends the cancel frame (when no terminal frame has arrived); the downlink iterator ends quietly. */
  dispose(): void
}

The handle represents one generation. When the carrier is lost, for await fails with RemoteStreamCarrierError and the handle ends there; whether to reopen, and with which arguments, belongs to the upper-layer protocol.
send is synchronous: it runs the isRemoteJsonValue check before enqueuing and throws synchronously on a value that is not lossless JSON. The browser's WebSocket.send has no write callback, so throttling sends would be meaningless; the Host-side inbox limit is the only backpressure point.
When a terminal frame (end / error) arrives, the mux client stops the pump and closes the uplink queue immediately on the receive path, so later send / end throw without waiting for the consumer's next read.
dispose() calls return() on the carrier iterator, so the cancel frame is sent even when nobody is reading the downlink; afterwards the handle's next() ends immediately and buffered items are dropped. A consumer that breaks out of for await early is equivalent to dispose().
The $stream supervisor, RemoteSnapshotStream, and RemoteJournalStream are upper-layer tools outside the transport layer. The object their open factories receive is the handle itself, with unchanged for await behavior; exposing send to these upper-layer protocols is deferred.

Wire frames

export type RemoteStreamClientMessage =
  | { readonly type: 'open'; readonly streamId: string; readonly endpoint: string; readonly payload: unknown }
  | { readonly type: 'item'; readonly streamId: string; readonly value?: unknown }   // new: uplink item
  | { readonly type: 'end'; readonly streamId: string }                              // new: uplink half-close
  | { readonly type: 'cancel'; readonly streamId: string }

/** Frames sent by the Host are unchanged. */
export type RemoteStreamServerMessage =
  | { readonly type: 'item'; readonly streamId: string; readonly value?: unknown }
  | { readonly type: 'error'; readonly streamId: string; readonly error: RemoteStreamFailure }
  | { readonly type: 'end'; readonly streamId: string }

Parsing rules: item has exactly the keys type and streamId plus an optional value, and value must be a lossless JSON value (isRemoteJsonValue); end has exactly the keys type and streamId. Any other frame is a frame-format violation, and the socket closes with 1008. No binary frames are introduced.

Logical stream state machine

Host side

                open frame
   (none) ───────────────────▶ opening ──── opener resolves ────▶ running
                                  │                                  │
                                  │ item frame: inbox.push           │ item frame: inbox.push
                                  │ end frame: inbox.end             │ end frame: inbox.end (uplink half-close)
                                  │                                  │ method yield: sends item frame
                                  │ cancel frame / socket close ────▶│ cancel frame / socket close: control.abort
                                  ▼                                  ▼
                               aborted ◀───────────────────────── finished (sends end or error frame)

item frames that arrive during opening enter the inbox: receive() is synchronous and creates the ActiveStream and its inbox while handling the open frame, whereas the opener is asynchronous. Items the Client sends immediately after open are not lost.

Client side

   call generated method ──▶ await socket ──▶ send open ──▶ handle usable
                                                            │ send(v): sends item; end(): sends end
                                                            │ downlink: inbox.next() → yield; end → ends; error → throws
                                                            │ downlink terminates first: send/end throw
                                                            │ dispose or caller signal abort: sends cancel (when no terminal frame has arrived)

Half-close and termination

Uplink	Downlink	Result
Client `end`	Still open	The Host's `uplink()` iteration ends; the method keeps producing. This is the stdin EOF form
Still open	The Host method finishes and sends `end`	The Client handle terminates, and later `send` / `end` throw; the Host calls `return()` on the `uplink` iterator and drops unconsumed items
Still open	The Host method throws and sends `error`	Same as above; the Client downlink fails with `RemoteError`
Client `dispose` or signal abort	Any	Sends `cancel`; the Host runs `control.abort()`, `cancellableStream` closes `uplink` first and then calls `return()` on the method iterator; no terminal frame is sent
Socket closes	Any	The Host aborts every stream and awaits `done`; on the Client every stream fails with `RemoteStreamCarrierError`
`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.

Backpressure

WebSocket has no per-stream window like HTTP/2: one connection is one serial write chain, and a slow stream holds up the other streams on the same connection. This property of the downlink is unchanged. For the uplink:

A bounded inbox plus whole-stream failure. The Host meters each logical stream's inbox in UTF-8 bytes of its frames, with the limit set by the Gateway config streamInboxBytes (default 262144) and applied to every stream. On overflow the whole logical stream fails with gateway/uplink-overflow: the downlink breaks with it, an error frame goes to the Client, and the uplink() iterator ends with the same error. No data is discarded; a method that does not read its uplink while the Client keeps sending is a usage error, and the failure exposes it explicitly.

Credit frames wait until a consumer that needs sustained high-volume uplink appears; the inbox limit is the initial credit, and adding them is incremental.

Security and validation

Item	Rule
Authentication	Same as today: `connection.requestRejection(req)` at upgrade; descriptor and lookup-parameter validation at open. Uplink items belong to an already authenticated stream and are not authenticated again
Peer	Admitted at upgrade as one `PeerScope`; `invocation.peer` of every stream on that socket is that Peer
Item validation	With an uplink codec, every item is decoded strictly, and a failure fails the whole stream with the existing `gateway/input-invalid` (details `{ endpoint, field: 'uplink' }`); without a codec, only JSON-safety validation applies
Frame size	A single frame keeps the `maxPayload` of `ws`; the inbox limit bounds the accumulated total

In-process and worker carriers

Carrier	Change
`ClientConnectionRpc.open` (`packages/client/connection/src/rpc.ts:245-250`) and `RpcStreamOpen` (`client/rpc.ts:21-25`)	Append an optional `uplink?: AsyncIterable<unknown>`; in process there are no frames and no inbox, the caller's iterator is the source of `uplink()` directly, and backpressure is the pace of the iterator itself
Worker tunnel (`packages/experimental/webworker-runtime`)	The page adds `stream-uplink-item` / `stream-uplink-end` frames; `serveStream` assembles them into an `uplink` handed to `seams.openStream`. Same-origin trusted boundary: no inbox limit, and items after `end` are dropped
`remote-mock` (`packages/test-support/remote-mock`)	`StreamHandle.uplink: AsyncIterable<unknown>`; `rpc.open` passes it through

An in-process carrier passes its Peer explicitly; the in-page Host of the worker tunnel passes connection.browserPeer.

Relation to `$events`

$events is a stream on a reserved endpoint; with uplink frames, waterfall answers can become uplink items on the same streamId, clientId becomes redundant because streamId already identifies the generation, and $events/result disappears together with the race at index.ts:527-529. Out of scope for this Note; see Deferred.

Type and wire definitions

typert protocol (`packages/typert/protocol/src`)

// types.ts
export type RemoteStream<Out, In = never> = AsyncIterable<Out> & { readonly [STREAM_UPLINK]?: In }
export type PeerId = Branded<'PeerId'>
export interface PeerScope { readonly id: PeerId; readonly ctx: Context; readonly principal: Principal; readonly signal: AbortSignal; dispose(): Promise<void> }
export interface RemoteInvocation { … }              // see Host face
declare module '@deepseek-ai/cordis' { interface Context { readonly invocation: RemoteInvocation | undefined } }

export interface InvocationDescriptor {
  // existing fields unchanged; mode still has only 'stream'
  readonly mode?: 'stream'
  /** Uplink item codec, generated from In of the return type RemoteStream<Out, In>; absent when In is never. */
  readonly uplink?: { readonly codec: TypertCodec }
  readonly cancellation?: { readonly parameter: 'signal' }
  readonly result: TypertCodec
}

The Remote decorator, RemoteMethodOptions, and RemoteMethodMarker recognize only mode: 'stream'.

typert generator (`packages/typert/generator/src`)

model.ts: InvocationModel.uplink?: { boundary: RemoteBoundaryModel }.
analyzer.ts remoteResultType: for mode: 'stream', the accepted return-type wrappers are Iterable<Out>, AsyncIterable<Out>, and RemoteStream<Out, In?>. RemoteStream is recognized the same way as the standard library's AsyncIterable: by symbol name plus declaring file (types.ts of @deepseek-ai/dsh-typert-protocol). The first type argument is the downlink item; when the second is present and is not never, an uplink boundary is generated under the key ${endpoint}:uplink.
emitter.ts: the descriptor literal emits uplink: { codec }; the generated Client signature returns RemoteStreamHandle<Out, In>.
The parameter loop recognizes no parameter named uplink.

Two names, one entry point

A Host method declares a stream with RemoteStream<Out, In>; the Client holds a RemoteStreamHandle<Out, In>. Both are exported from the main entry point of @deepseek-ai/dsh-typert-protocol, and the handle interface has no Host dependency. The generated Client contract writes the return type as RemoteStreamHandle<Out, In>. One name has one meaning: the RemoteStream that Client code obtains from the main entry point is always the declaration type and is never confused with the handle.

Gateway Host (`packages/api/gateway/src`)

// types.ts
export interface InvokeRemoteRequest {
  readonly namespace: string
  readonly method: string
  readonly args: Readonly<Record<string, unknown>>
  /** Uplink items; absent is equivalent to an iterator that ends immediately. */
  readonly uplink?: AsyncIterable<unknown>
  /** The Peer that initiated the call; required on every path. */
  readonly peer: PeerScope
  readonly signal?: AbortSignal
}

export interface TypertGatewayWireStream {
  open: (endpoint: string, payload: unknown, uplink: AsyncIterable<unknown>, peer: PeerScope, signal: AbortSignal) => Promise<AsyncIterable<unknown>>
  failure: (error: unknown) => RemoteStreamFailure
}

// stream-server.ts
export type RemoteStreamOpener = (
  endpoint: string,
  payload: unknown,
  uplink: AsyncIterable<unknown>,
  peer: PeerScope,
  control: AbortController,
  window: RemoteUplinkWindow,
) => Promise<AsyncIterable<unknown>>

// index.ts Config addition
/** Uplink inbox limit per logical stream, in UTF-8 bytes of frames (default 262144). */
readonly streamInboxBytes?: number

prepareInvocation:

const args = await Promise.all(descriptor.parameters.map(parameter => this.resolveParameter(parameter, request.args, endpoint)))
const signal = request.signal ?? NEVER_ABORTED_SIGNAL
const invocation = new GatewayInvocation(
  { namespace: request.namespace, method: request.method, args: request.args },
  descriptor.service,
  request.peer,
  signal,
  new UplinkDecoder(request.uplink ?? EMPTY_ASYNC_ITERABLE, descriptor.uplink?.codec, endpoint, control),
)
const callReceiver = receiverContext.extend({ invocation }).get(descriptor.service) as object
if (descriptor.cancellation !== undefined) args.push(signal)

GatewayInvocation.uplink() returns the decoder on the first call and throws on the second. UplinkDecoder is a hand-written iterator: with a codec it runs decode(codec, value, endpoint, 'uplink') per item and, on failure, calls control.abort(failure) and throws; without a codec it applies only the isRemoteJsonValue check. The finally of cancellableStream closes the decoder first (invocation.close()) and then awaits the method iterator's return(), so a method blocked on uplink.next() can exit. The mode branches are the existing two, unary and stream. SRC fallback: the descriptor has no uplink codec, and uplink() delivers unknown.

Gateway Host mux (`stream-server.ts`)

interface ActiveStream { readonly control: AbortController; readonly inbox: UplinkInbox; done: Promise<void> }

/** Bounded uplink queue; iterated as the source of uplink(). */
class UplinkInbox implements AsyncIterable<unknown>, AsyncIterator<unknown> {
  constructor(maxBytes: number, onOverflow: (error: RemoteError) => void)
  push(value: unknown, frameBytes: number): void   // push after ended fails the stream with gateway/protocol; overflow calls onOverflow
  end(): void                                        // idempotent
  fail(error: unknown): void                         // ends the iterator with an error on cancel or carrier closure
  next() / return()                                  // hand-written: next races failure, return releases immediately
}

receive(text) dispatch:

open   : exists → throw (socket 1008); otherwise create { control, inbox, done } and start the pump
item   : missing → ignore; otherwise inbox.push(value, bytes)
end    : missing → ignore; otherwise inbox.end()
cancel : missing → ignore; otherwise control.abort(new Error('Remote stream cancelled'))

pump passes the inbox and the socket's Peer to the opener and calls inbox.fail(...) in its finally. An abort whose reason is a RemoteError is sent by the pump as an error frame (a failure raised by the Gateway or the mux); an abort whose reason is a plain Error (cancel, socket closure) sends no terminal frame.

Gateway Client (`packages/api/gateway/src/client`)

prepareInvocation strips no uplink parameter; the positional parameters are the business parameters plus the optional signal.
invokeStream returns a RemoteStreamHandle: it holds an uplink queue internally, send enqueues and a pump sends the item frames, end sends the end frame, and dispose aborts the generation signal. The downlink is still the iterator that openRemoteStream returns.
RemoteStreamMuxClient.open(endpoint, payload, signal, uplink?): after sending the open frame it starts the uplink pump, which sends one item per item and end when the source ends; when the downlink terminates or the signal aborts, it stops the pump and calls the source's return() (without awaiting, so a keyboard-style generator cannot hang it); when the source throws, it calls inbox.fail(error), the downlink fails with that error, and finally sends cancel under the existing rules. The pump checks this.socket === socket, so an old generation never sends frames on a new socket.
In-process carrier: connection.rpc.open(channel, endpoint, payload, signal, uplink).

Placement

Package	Carries
`dsh-typert-protocol` (`types.ts`, `index.ts`)	`RemoteStream`, `RemoteStreamHandle`, `PeerId`, `PeerScope`, `RemoteInvocation`, the `ctx.invocation` declaration merge, `InvocationDescriptor.uplink`; the decorator recognizes only `mode: 'stream'`; `TypertRemoteService` registers the `invocation` accessor when constructed
`dsh-typert-registry`	Load-time validation that `uplink.codec` is a valid strict codec
`dsh-typert-generator` (`model.ts`, `analyzer.ts`, `emitter.ts`)	`remoteResultType` recognizes `RemoteStream<Out, In>` and generates the `uplink` boundary; the descriptor emits `uplink: { codec }`; the generated Client signature returns `RemoteStreamHandle<Out, In>`; the parameter loop accepts only business parameters and a final `signal`
`dsh-client-connection` (`peer.ts`, `rpc.ts`, `rpc-host.ts`, `index.ts`)	The `Peer` class; `connection.browserPeer` and `admit`; the `/api` and upgrade routes obtain the Peer through `admit`; the optional `uplink` of `rpc.open`
`dsh-api-gateway` Host (`stream-protocol.ts`, `stream-server.ts`, `types.ts`, `index.ts`)	`item` / `end` frames; `UplinkInbox`; `Config.streamInboxBytes`; `GatewayInvocation` and `extend({ invocation })`; `UplinkDecoder`; `handleUpgrade(req, socket, head, peer)`
`dsh-api-gateway` Client (`client/index.ts`, `client/stream-client.ts`)	`invokeStream` returns the handle; the uplink pump is driven by the handle's queue
`dsh-webworker-runtime`, `dsh-remote-mock`	The worker tunnel's two uplink frames and the `serveStream` assembly; `StreamHandle.uplink`; the direct proxy returns the real handle
Documentation	The READMEs of gateway, the api group, connection, protocol, and generator; `docs/api-gateway`; the type-equiv blocks of `docs/subsystems/typert`; the config and Cordis catalogs

Key sequences

Open, uplink, half-close, downlink end

Client caller      mux client              mux server              gateway                 Host method
   │ attach(req)        │                       │                     │                         │
   │───────────────────▶│ open { id, ep, payload } ──────────────────▶│ create inbox            │
   │ s.send(v1)         │ item { id, v1 } ───────────────────────────▶│ inbox.push(v1)          │
   │                    │                       │                     │ prepareInvocation       │
   │                    │                       │                     │ extend({ invocation })  │
   │                    │                       │                     │── method(req, signal) ─▶│
   │                    │                       │                     │                         │ uplink() → v1
   │ s.send(v2)         │ item { id, v2 } ───────────────────────────▶│ inbox.push(v2)          │ → v2
   │ s.end()            │ end { id } ────────────────────────────────▶│ inbox.end()             │ iteration ends
   │◀── yield o1 ───────│◀────────────────────── item { id, o1 } ─────│◀── yield o1 ────────────│
   │◀── ends ───────────│◀────────────────────── end { id } ──────────│◀── method returns ──────│ gateway calls uplink.return()

Client dispose

Client caller      mux client              mux server              gateway                 Host method
   │ s.dispose()        │                       │                     │                         │
   │───────────────────▶│ stop pump; source return() (not awaited)    │                         │
   │◀── iterator ends ──│ cancel { id } ─────────────────────────────▶│ control.abort()         │
   │                    │                       │                     │ cancellableStream: invocation.close() closes uplink
   │                    │                       │                     │ iterator.return() ─────▶│ finally cleanup
   │                    │                       │ no terminal frame   │                         │

inbox overflow

Client caller      mux client              mux server                        gateway / Host method
   │ s.send(vN)         │ item { id, vN } ─────▶│ inbox.push: bytes > streamInboxBytes │
   │                    │                       │ onOverflow → control.abort(RemoteError uplink-overflow)
   │                    │                       │                                      │ uplink() ends with that error; method iterator return()
   │◀── throw ──────────│◀──────────────────────│ error { id, uplink-overflow }: the pump sends an error frame because the reason is a RemoteError

Boundary matrix

Event	opening	running	Uplink ended	Downlink terminated
`item` received	Buffered in the inbox	Into the inbox	The stream fails with `gateway/protocol`	Ignored
`end` received	inbox.end	inbox.end	Idempotent	Ignored
`cancel` received	abort	abort	abort	Ignored
inbox overflow	The stream fails with `uplink-overflow`	Same	Impossible	Impossible
Decode failure	Impossible	The stream fails with `input-invalid`	Same	Impossible
Host method finishes	Impossible	Sends `end`; `uplink.return()`	Sends `end`	Already sent
Host method throws	Sends `error`	Sends `error`; `uplink.return()`	Sends `error`	Already sent
Socket closes	abort, await done	Same	Same	Nothing
Client `send`	Queued, sent after open	Sent	Throws	Throws
Client `end`	Recorded, sent after open	Sent	Idempotent	Ignored
Client-side downlink ends first	Impossible	Stop the pump; source `return()`; no `end` sent	Nothing	Nothing

Error codes

Code	When
`gateway/input-invalid`	Existing: a parameter codec fails to decode; new: an uplink item fails to decode
`gateway/uplink-overflow`	New: the inbox exceeds `streamInboxBytes`
`gateway/protocol`	New: an `item` after `end`
`gateway/cancelled`	Existing: `signal` aborted

New codes are registered in packages/api/gateway/src/remote-error-codes.ts and in the README's error-code section.

Minimal echo method

packages/api/gateway/tests/echo-stream.host.spec.ts runs the same method through the WebSocket carrier and through the in-process carrier:

class EchoService extends TypertRemoteService {
  constructor(ctx: Context) { super(ctx, 'echo', { namespace: 'echo' }) }

  @Remote({ mode: 'stream' })
  async *echo(prefix: string, signal: AbortSignal): RemoteStream<string, string> {
    for await (const item of this.ctx.invocation!.uplink<string>()) {
      signal.throwIfAborted()
      yield `${prefix}${item}`
    }
  }
}

// Client
const stream = remote.echo.echo('> ')
stream.send('a'); stream.send('b'); stream.end()
const replies: string[] = []
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.

Passing a schema at runtime: uplink(schema). Bypasses the generator, the Client loses type projection, and it splits from the "strict codec per item" system. The codec still comes from the generator; the generic in the method is only an assertion.

Having the analyzer cross-check the hand-written generic of uplink<In>() against the return type. A business-specific check whose mistakes the caller bears, consistent with the other hand-written assertions in the repository; not added.

A transport-layer handle with reopening, cursors, and a reopen hook. Reopening after disconnection belongs to the upper-layer protocol: a job resumes through an opened frame carrying from and a client that remembers next and calls again; snapshot and journal each have their own anchors and classification. The transport layer knows one generation.

Putting the Client's send on the AsyncIterable the generated method returns. Hanging methods on an AsyncIterable is unconventional; the handle type RemoteStreamHandle extends it explicitly.

Copying the Access mechanism. Access control is business semantics, served by vouch and the remote/invoke waterfall; this Note only needs "whose stream is this", and PeerScope suffices. Method-level access control later moved to Peer principals and method-level Remote access control, which supersedes this part.

Keeping unary RPC for the uplink. One HTTP round trip, one authentication, and one lookup per uplink; two carriers with no ordering relation that only Client-side serialization can order; no EOF; cancellation that does not cover both halves; a hand-made correlation key per feature.

A native HTTP request body stream. A one-shot request body with no association to the downlink stream, bypassing Typert descriptors, and browser support for full-duplex fetch cannot be relied on.

One dedicated WebSocket per duplex stream. Loses the authentication, heartbeat, and multiplexing shared with the existing downlink streams, and every feature manages its own socket lifecycle.

WebTransport or HTTP/2 bidirectional streams. They have per-stream flow control, but support on the browser and Node sides, proxy traversal, and the Electron carrier are all immature; adding two frames to the existing mux is orders of magnitude cheaper than replacing the transport layer.

Consequences

Bought: one stream with two directions, so interactive scenarios such as terminal keystrokes, approval answers, and background-job stdin no longer each build their own "one stream plus one unary"; a Host method knows its caller; the uplink and downlink types are declared in one place, the return type; uplink items come from the browser and are validated strictly per item at the Host, while downlink items are Host-produced typed values that pass straight through; the generated output and wire frames of existing AsyncIterable<Out> methods and unary methods are completely unchanged.
this.ctx.invocation is undefined outside a Remote call, so a method that reads it must handle the optional; a service method invoked only directly in process correctly reads undefined.
Web terminals use attach for screen output and input commands, with provider acknowledgements and per-attachment writable state; their Client retains the current generation's handle and never replays input.
The inbox limit applies to every stream: a method that does not read its uplink fails as a whole when it receives many uplink frames. This is a deliberate explicit failure, recorded in the README.
Every browser connection shares browserPeer, and admit is the only network admission point; in-process code creates further Peers with connection.createPeer().
No cross-direction ordering guarantee between uplink and downlink; a protocol that needs request-response pairing carries its own sequence numbers.
Upper-layer supervisors expose only the downlink: $stream, snapshot and journal do not expose send; a domain that writes uplink retains its generated physical handle for that generation.
The worker tunnel behaves differently from the WebSocket carrier: no inbox limit, and items after end are dropped, because page to worker is a same-origin trusted boundary.

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.
typert: the analyzer's handling of RemoteStream<Out, In>, RemoteStream<Out>, never, and the existing AsyncIterable<Out>; decorator options; registry load-time validation.
connection: the three outcomes of admit, 401, 403, and browserPeer; browserPeer reaching handlers through the bridge and the shared handler; release with the Connection.
remote-mock and the worker tunnel: uplink pass-through; the direct proxy returning the real handle.
Existing stream and unary method tests pass without changed expectations.

Deferred

$stream / snapshot / journal expose the handle's send to upper-layer protocols.
Duplex $events: waterfall answers become uplink items on the same streamId, and $events/result and clientId retire.
Credit frames: wait for a consumer that needs sustained high-volume uplink; the inbox limit is the initial credit.
A Host-initiated half-close frame (the Host stops reading the uplink but keeps writing the downlink): no consumer; not added.
uplink() on unary methods: it can read items that arrive while the method runs, semantically valid but of doubtful use; allowed, noted in the README.

Remote event delivery: the unary $events/result answer path; the duplex $events item under Deferred partially supersedes it once it lands.
Session history and event transport: the $stream() supervisor and the journal / snapshot protocols above it, which this note leaves unchanged.
Web sidebar terminal: the attach stream carries input commands, provider acknowledgements and per-attachment writable state; reconnect never replays input.

中文

问题

现状

dsh-api-gateway 在 /api/remote.mux 的一条 WebSocket 上复用 Typert Remote 流。Host 方法通过 @Remote({ mode: 'stream' }) 和 RemoteStream<Out, In> 声明两个方向；生成的 Client 方法返回 RemoteStreamHandle<Out, In>。既有的 Iterable 和 AsyncIterable 返回类型声明仅下行的方法。可选的末位 signal: AbortSignal 由 Gateway 提供，不进入线路参数。

两个方向共用一个 streamId：

方向	帧	定义
Client 到 Host	`{ type: 'open', streamId, endpoint, payload }`	`stream-protocol.ts`
Client 到 Host	`{ type: 'item', streamId, value? }`	`stream-protocol.ts`
Client 到 Host	`{ type: 'end', streamId }`	`stream-protocol.ts`
Client 到 Host	`{ type: 'cancel', streamId }`	`stream-protocol.ts`
Host 到 Client	`{ type: 'item', streamId, value? }`	`stream-protocol.ts`
Host 到 Client	`{ type: 'error', streamId, error: { code, message, details } }`	`stream-protocol.ts`
Host 到 Client	`{ type: 'end', streamId }`	`stream-protocol.ts`

Client 句柄的 send() 提供上行项，end() 半关闭上行，dispose() 取消逻辑流。Host 通过 this.ctx.invocation.uplink<In>() 读取解码后的项；调用上下文同时携带获准接入的 Peer 和取消信号。

传输层负责校验、取消和有界 inbox。消费者协议负责输入权限、确认以及断线后的恢复。

上行消费者

功能	上行方式	位置	代价
Web 终端输入与尺寸调整	`terminal.attach` 承载上行 `TerminalInput` 和下行 `TerminalFrame`；Client 连续发送命令，并在收到 `ack` 后释放原始输入字节预算	`packages/api/terminal-controller/src/client/model.ts`；Host `src/terminal.ts`	当前附件拥有输入权；Client 预算耗尽时报 `inputFull`；重连后绝不重放输入
文件上传	`Blob` 或 `ReadableStream` 走 Remote 层之外的原生 `POST /api/session/uploadFileBinary`，`duplex: 'half'`	`packages/client/file-upload/src/client/runtime.ts`；路由 `src/index.ts`	不经 Typert 描述符，没有类型投影
审批与提问的回答	Host 在 `$events` 流上下发 waterfall 帧，Client 用一个独立的 unary RPC `$events/result` 回答，靠 `clientId` 加 `eventId` 关联	`packages/api/gateway/src/client/remote-events.ts`；Host `index.ts`	第二套关联键；`index.ts` 记录了由此产生的竞态

这些消费者都需要在相关操作仍进行时从 Client 向 Host 传送数据。Web 终端使用双工流；文件上传和事件回答保留各自的传输方式。

消费者职责

输入权限。终端通过 Host 本地的 follower 持有控制附件。快照和状态帧报告各附件的 writable 状态；控制权转移后，旧附件继续接收输出。
顺序与确认。终端输入和尺寸调整共用连续命令序列及终端操作队列。ack 报告提供者已接受操作或输入权限拒绝，绝不授权断线后重放。
独立生命周期。取消 attach 会停止其输入和输出传递、丢弃尚未开始的命令，并等待在途输入操作收尾。PTY 的生命周期由保留机制和显式关闭独立管理。
协调字节限额。Gateway 通过 invocation.uplinkWindow 提供完整帧 inbox 容量及本流的 JSON 封装开销。终端在快照中公布窗口，并通过提供者确认释放在途额度。窗口满时，输入和尺寸调整命令在有界 Client 队列中等待。maxInputBytes 限制原始 UTF-8 输入；maxBufferedBytes 分别限制 Client 保留的完整编码输入及每个 follower 保留的输出。
仍独立的载体。审批回答仍通过 clientId 和 eventId 将 $events 与 $events/result RPC 关联。将该协议迁入双工流仍属后续工作。

与后台 job 的关系

后台 job 的进程形双工流面（已存档为草稿 PR，分支 worktree/job-stdio-stream-pair-archive）把 job 建成一对镜像的流面，浏览器要接它就需要一条既下行又上行的 Remote 流。它是本 Note 的动机之一，但不是本 Note 的范围：job 要不要一种专用的读写流，是 jobs 那条线自己的架构问题。本 Note 只覆盖传输层。

目标

传输层只有一种流，天生双工：下行 item / end / error，上行 item / end，一个 streamId，一个代际。
上行类型与下行类型写在同一个位置：方法的返回类型。
Host 方法通过调用上下文取上行；调用上下文同时携带调用者（Peer）。
客户端拿到的不再是裸 AsyncIterable，而是一个能读、能写、能关的句柄。
上层协议（snapshot、journal、任何业务自定义的流）不进传输层。

决定

一句话

所有 Remote 流都允许客户端上行。方法的返回类型 RemoteStream<Out, In = never> 同时声明下行项与上行项的类型；生成器从中生成两个 codec。Host 方法通过 this.ctx.invocation.uplink<In>() 取上行迭代器，通过 this.ctx.invocation.peer 知道调用者。客户端生成方法返回 RemoteStreamHandle<Out, In>：for await 读下行，send / end 写上行，dispose 关。断线重开、游标续传、基线校验都是上层协议的事。

                 Client                                             Host
   ─────────────────────────────────              ───────────────────────────────────────────
   const s = remote.job.attach(req, signal)
        │  open { streamId, endpoint, payload } ───────▶  opener(endpoint, payload, uplink, peer, control, window)
        │                                                     │ prepareInvocation
        │  s.send(v1) → item { streamId, v1 }    ───────▶  inbox.push(v1)   │ invocation = { request, service, peer, signal, uplink() }
        │  s.send(v2) → item { streamId, v2 }    ───────▶  inbox.push(v2)   │ receiverContext.extend({ invocation }).get(service)
        │  s.end()    → end  { streamId }        ───────▶  inbox.end()      │ method(...args, signal)
        │                                        ◀───────  item { streamId, o1 }   │   for await (v of this.ctx.invocation.uplink()) …
        │  for await (o of s) …                  ◀───────  item { streamId, o2 }   │   yield o …
        │  s.dispose() → cancel { streamId }     ───────▶  control.abort()         │ signal aborts; uplink.return()
        │                                        ◀───────  end | error

术语

术语	含义
逻辑流	一个 `streamId` 标识的一次调用，跨越 `open` 到终止帧
代际	一条逻辑流的一次物理生命期。传输层只认一个代际；断线即失败，重开是上层的事
下行	Host 到 Client 的 `item` 序列，以 `end` 或 `error` 终止
上行	Client 到 Host 的 `item` 序列，以 `end` 终止（半关闭），或随 `cancel` 一起结束
半关闭	一个方向已结束而另一个方向仍开放
项	一个方向上的一个业务值；每项独立经 codec 校验
inbox	Host 侧每逻辑流一个的有界上行队列
Peer	一条连接代表的一方，一个携带 principal 的 `PeerScope`；所有浏览器连接共享 `connection.browserPeer`，进程内代码经 `connection.createPeer()` 自建（Peer 身份属性）
调用上下文	`RemoteInvocation`：本次调用的请求、接收服务、Peer、取消信号与上行入口

类型签名

// @deepseek-ai/dsh-typert-protocol
/**
 * One Remote stream. Host face: the method returns it, and at runtime it is AsyncIterable<Out>.
 * On the Client face the generated method returns RemoteStreamHandle<Out, In>; each name has exactly one meaning.
 * In is the item type the Client may send uplink; the default never means the method reads no uplink.
 */
export type RemoteStream<Out, In = never> = AsyncIterable<Out> & { readonly [STREAM_UPLINK]?: In }

// Host
@Remote({ mode: 'stream' })
async *attach(request: JobAttachRequest, signal: AbortSignal): RemoteStream<JobFollowFrame, JobInputFrame> {
  const uplink = this.ctx.invocation.uplink<JobInputFrame>()
  void (async () => { for await (const frame of uplink) … })()
  yield …
}

// Generated Client signature: same parameters, returns the handle
attach(request: JobAttachRequest, signal?: AbortSignal): RemoteStreamHandle<JobFollowFrame, JobInputFrame>

// Client usage
const stream = remote.job.attach(req, signal)
for await (const frame of stream) …
stream.send(frame)     // typed as JobInputFrame
stream.end()
stream.dispose()

RemoteStream 是别名而不是接口，是刻意的：将来若把传输层做成 Node 流形态，只换别名的定义与两端的适配，方法签名、描述符、上层协议都不动。

Host 面

/** The context of this Remote call. A method reads it through this.ctx.invocation. */
export interface RemoteInvocation {
  readonly request: {
    readonly namespace: string
    readonly method: string
    readonly args: Readonly<Record<string, unknown>>
  }
  /** Cordis service key of the receiving service. */
  readonly service: string
  /** The Peer that initiated the call; every carrier names it explicitly. */
  readonly peer: PeerScope
  /** Carrier cancellation: Client cancel, socket closure, uplink failure. */
  readonly signal: AbortSignal
  readonly uplinkWindow?: RemoteUplinkWindow | undefined
  /**
   * The uplink items of this call. Can be taken once; the second call throws. With an uplink codec
   * in the descriptor each item is decoded to In; without one, items are delivered as unknown with
   * only a JSON-safety check.
   * The iteration ends after the Client's end; when the method ends the downlink, the Gateway calls
   * its return() and drops unconsumed items.
   * The generic In is only the caller's type assertion; runtime decodes per the descriptor and does not cross-check.
   */
  uplink<In = unknown>(): AsyncIterable<In>
}

declare module '@deepseek-ai/cordis' {
  interface Context {
    /** The Remote call this Context was derived for; undefined on a Context not derived from a Remote call. */
    readonly invocation: RemoteInvocation | undefined
  }
}

规则：

任何 mode: 'stream' 方法都可以调用 this.ctx.invocation.uplink()；unary 方法也可以，但它的流在方法返回时结束，上行项只在方法运行期间可读。
uplink() 未被调用时，上行项在 inbox 里攒着，直到超限（见背压）或流结束被丢弃。这是刻意的：不读上行的方法不用知道上行存在。
In 缺省为 never 时描述符没有 uplink codec；客户端句柄的 send 类型为 never，编译期就不能发。运行时若仍有帧到达（手工构造的客户端），按 unknown 交付。
this.ctx.invocation 在非 Remote 调用的 Context 上是 undefined；方法内读它时按 TS 的可选处理，或在方法第一行断言。该属性的 accessor 由 TypertRemoteService 基类构造时注册一次，任何装有 Remote 服务的组合都能读它。

调用上下文如何到达方法

机制依赖 Cordis 的两条既有行为：

ctx.extend(meta) 是 Object.create(parent) 加自有属性（vendor/cordis/src/context.ts:99-107），无注册、无 fiber、不需释放，每次调用派生一个是零成本的。
ctx.get(service) 返回 createTraceable(ctx, service) 代理（vendor/cordis/src/reflect.ts:233-234，utils.ts:117-125）。读 ctx 属性直接返回访问方的 Context（utils.ts:175）；读方法时用 createShadowMethod 包一层，apply 时把 this 换成 shadow receiver，shadow 上的 ctx 是 ctx.extend({ [symbols.shadow]: origin })（utils.ts:149-163）。
网关因此写 receiverContext.extend({ invocation }).get(descriptor.service) 取一个调用视图，Reflect.apply(method, callReceiver, args)；方法体里 this.ctx.invocation 沿原型链读到本次调用。async generator 体延后执行不受影响，因为 this 在调用时已绑定为 shadow 对象。

PeerScope

/** The opaque identity of one Peer. */
export type PeerId = Branded<'PeerId'>

/** One Peer's session on this Host: opened and released by its creator; ctx owns connection-lifetime registrations. */
export interface PeerScope {
  readonly id: PeerId
  readonly ctx: Context
  /** Caller identity fixed at creation. */
  readonly principal: Principal
  /** Aborted once the Peer is disposed. */
  readonly signal: AbortSignal
  /** Participation in forwarded interactions; independent of RBAC. */
  readonly interactions: 'answer' | 'none'
  dispose(): Promise<void>
}

@deepseek-ai/dsh-client-connection 实现唯一的 Peer 类 Peer(owner, { principal, interactions? })，用 createScope(owner, peer) 建立，与 Agent 建自己的 scope 是同一机制：Peer 对象就是 ScopeKey，peer.ctx 承接连接期效果，scope 结束时 peer.signal 中止，dispose() 让 fiber 静默。Connection 在服务 apply 时以 owner principal 创建 connection.browserPeer；principal、createPeer() 与访问检查归 Peer 身份属性与 Remote 方法级访问控制。

成员	语义
`connection.browserPeer: PeerScope`	所有通过认证的浏览器连接共享的 Peer；principal 为 owner；随 connection 生灭
`connection.admit(request): PeerAdmission`	检查 Host/Origin 与浏览器或 API 签名 Cookie；返回 `{ rejection: 401 \| 403 }` 或凭据对应的存活 Peer；API Peer 在 connect 时发布，并由其生命周期 socket 负责释放

网关 handleUpgrade(req, socket, head, peer) 为该 socket 绑一个固定 Peer 的 opener，并在 peer.ctx.effect 里登记 socket 关闭；scope 已失效时改为直接以 1001 关闭 socket。InvokeRemoteRequest.peer、RemoteStreamOpener 与 wireStream.open 的 peer 形参均为 required，每个进程内调用方都显式给出 Peer，网关不自建 Peer。按 id 查 Peer、Peer 的开关事件，都等第一个需要它们的消费者出现时再加。

Peer 携带自己的 principal；该 principal 能调用什么，由 Peer 身份属性与 Remote 方法级访问控制定义的 role、binding 与 policy 判定。

Client 面

/** The handle a generated method returns on the Client face. A different name from the Host's RemoteStream, exported from the same entry point. */
export interface RemoteStreamHandle<Out, In> extends AsyncIterable<Out> {
  /** Send one uplink item. Throws on a value that is not lossless JSON, after termination, or after end. */
  send(item: In): void
  /** Half-close the uplink: sends the end frame. Idempotent. */
  end(): void
  /** Cancel the whole logical stream: sends the cancel frame (when no terminal frame has arrived); the downlink iterator ends quietly. */
  dispose(): void
}

句柄代表一个代际。载体丢失时 for await 以 RemoteStreamCarrierError 失败，句柄到此结束；要不要重开、带什么参数重开，是上层协议的事。
send 是同步的：入队前做 isRemoteJsonValue 校验，非无损 JSON 值同步抛错。浏览器 WebSocket.send 没有写回调，发送节流没有意义；Host 侧的 inbox 上限是唯一的背压点。
收到终止帧（end / error）时，mux 客户端在收帧路径上立即停泵并关闭上行队列，之后 send / end 抛错，不等消费者下一次读。
dispose() 调用载体迭代器的 return()，让 cancel 帧在无人读下行时也发出；之后句柄的 next() 直接结束，缓冲项丢弃。消费者提前 break 出 for await 等价于 dispose()。
$stream 监督器、RemoteSnapshotStream、RemoteJournalStream 是上层工具，不在传输层内。它们的 open 工厂拿到的对象现在就是句柄，for await 行为不变；把 send 透出到这些上层协议是后续的事。

线路帧

export type RemoteStreamClientMessage =
  | { readonly type: 'open'; readonly streamId: string; readonly endpoint: string; readonly payload: unknown }
  | { readonly type: 'item'; readonly streamId: string; readonly value?: unknown }   // new: uplink item
  | { readonly type: 'end'; readonly streamId: string }                              // new: uplink half-close
  | { readonly type: 'cancel'; readonly streamId: string }

/** Frames sent by the Host are unchanged. */
export type RemoteStreamServerMessage =
  | { readonly type: 'item'; readonly streamId: string; readonly value?: unknown }
  | { readonly type: 'error'; readonly streamId: string; readonly error: RemoteStreamFailure }
  | { readonly type: 'end'; readonly streamId: string }

解析规则：item 精确键 type streamId 加可选 value，value 须为无损 JSON 值（isRemoteJsonValue）；end 精确键 type streamId。其余形状是帧格式违规，socket 以 1008 关闭。不引入二进制帧。

逻辑流状态机

Host 侧

                open frame
   (none) ───────────────────▶ opening ──── opener resolves ────▶ running
                                  │                                  │
                                  │ item frame: inbox.push           │ item frame: inbox.push
                                  │ end frame: inbox.end             │ end frame: inbox.end (uplink half-close)
                                  │                                  │ method yield: sends item frame
                                  │ cancel frame / socket close ────▶│ cancel frame / socket close: control.abort
                                  ▼                                  ▼
                               aborted ◀───────────────────────── finished (sends end or error frame)

opening 期间到达的 item 帧进入 inbox：receive() 是同步的，open 帧处理时就创建 ActiveStream 与 inbox，而 opener 是异步的。客户端在 open 之后立刻发的项不会丢。

Client 侧

   call generated method ──▶ await socket ──▶ send open ──▶ handle usable
                                                            │ send(v): sends item; end(): sends end
                                                            │ downlink: inbox.next() → yield; end → ends; error → throws
                                                            │ downlink terminates first: send/end throw
                                                            │ dispose or caller signal abort: sends cancel (when no terminal frame has arrived)

半关闭与结束语义

上行	下行	结果
客户端 `end`	仍开放	Host 的 `uplink()` 迭代结束；方法继续产出。这是 stdin EOF 的形状
仍开放	Host 方法结束，发 `end`	客户端句柄终止，之后 `send` / `end` 抛错；Host 调用 `uplink` 迭代器的 `return()`，未消费项丢弃
仍开放	Host 方法抛错，发 `error`	同上；客户端下行以 `RemoteError` 失败
客户端 `dispose` 或 signal 中止	任意	发 `cancel`；Host `control.abort()`，`cancellableStream` 先关 `uplink`，再调方法迭代器的 `return()`；不发终止帧
socket 关闭	任意	Host 中止全部流并等待 `done`；客户端每条流以 `RemoteStreamCarrierError` 失败
`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 的顺序。上行与下行之间没有跨方向顺序保证；需要请求应答配对的协议自带序号。

背压

WebSocket 没有 HTTP/2 那样的每流窗口，一个连接一条串行写链，慢流会拖住同连接的其他流。下行的这一性质不变。上行：

有界 inbox 加整流失败。 Host 每条逻辑流的 inbox 按帧的 UTF-8 字节计量，上限为网关配置 streamInboxBytes（默认 262144），对所有流生效。超限时整条逻辑流以 gateway/uplink-overflow 失败：下行一起断，向客户端发 error 帧，uplink() 迭代器以同一错误结束。数据不丢弃；一个方法不读上行而客户端持续发，是用法错误，由失败显式暴露。

credit 帧留待需要持续大流量上行的消费者出现时再加；inbox 上限就是初始授信，加法是增量。

安全与校验

项	规则
鉴权	与今天相同：升级时 `connection.requestRejection(req)`；打开时描述符与查找参数校验。上行项属于一条已鉴权的流，不重复鉴权
Peer	升级时接纳为一个 `PeerScope`，该 socket 上所有流的 `invocation.peer` 都是它
项校验	有 uplink codec 时逐项严格解码，失败即整条流以既有的 `gateway/input-invalid` 失败（details `{ endpoint, field: 'uplink' }`）；无 codec 时只做 JSON 安全校验
帧大小	单帧沿用 `ws` 的 `maxPayload`；inbox 上限限制累计量

进程内与 worker 载体

载体	改为
`ClientConnectionRpc.open`（`packages/client/connection/src/rpc.ts:245-250`）与 `RpcStreamOpen`（`client/rpc.ts:21-25`）	追加可选 `uplink?: AsyncIterable<unknown>`；进程内没有帧也没有 inbox，调用方迭代器直接作为 `uplink()` 的源，背压就是迭代器本身的节拍
worker 隧道（`packages/experimental/webworker-runtime`）	页面新增 `stream-uplink-item` / `stream-uplink-end` 帧；`serveStream` 把它们组装成 `uplink` 交给 `seams.openStream`。同源可信边界：没有 inbox 上限，`end` 之后的项丢弃
`remote-mock`（`packages/test-support/remote-mock`）	`StreamHandle.uplink: AsyncIterable<unknown>`；`rpc.open` 透传

进程内载体显式传入 Peer；worker 隧道的页内 Host 传 connection.browserPeer。

与 `$events` 的关系

$events 是一条保留端点的流；有了上行帧，waterfall 的回答可以成为同一 streamId 上的上行项，clientId 因 streamId 已标识代际而多余，$events/result 与 index.ts:527-529 的竞态一起消失。不在本 Note 范围，见「后续」。

类型与线路定义

typert protocol（`packages/typert/protocol/src`）

// types.ts
export type RemoteStream<Out, In = never> = AsyncIterable<Out> & { readonly [STREAM_UPLINK]?: In }
export type PeerId = Branded<'PeerId'>
export interface PeerScope { readonly id: PeerId; readonly ctx: Context; readonly principal: Principal; readonly signal: AbortSignal; dispose(): Promise<void> }
export interface RemoteInvocation { … }              // see Host face
declare module '@deepseek-ai/cordis' { interface Context { readonly invocation: RemoteInvocation | undefined } }

export interface InvocationDescriptor {
  // existing fields unchanged; mode still has only 'stream'
  readonly mode?: 'stream'
  /** Uplink item codec, generated from In of the return type RemoteStream<Out, In>; absent when In is never. */
  readonly uplink?: { readonly codec: TypertCodec }
  readonly cancellation?: { readonly parameter: 'signal' }
  readonly result: TypertCodec
}

Remote 修饰器、RemoteMethodOptions、RemoteMethodMarker 恢复为只认 mode: 'stream'。

typert generator（`packages/typert/generator/src`）

model.ts：InvocationModel.uplink?: { boundary: RemoteBoundaryModel }。
analyzer.ts remoteResultType：对 mode: 'stream'，接受的返回类型包装器为 Iterable<Out>、AsyncIterable<Out>、RemoteStream<Out, In?>。识别 RemoteStream 的方式与识别标准库 AsyncIterable 相同：符号名加声明所在文件（@deepseek-ai/dsh-typert-protocol 的 types.ts）。第一个类型参数是下行项，第二个存在且不是 never 时生成 uplink boundary，键名 ${endpoint}:uplink。
emitter.ts：描述符字面量输出 uplink: { codec }；生成的 Client 签名返回 RemoteStreamHandle<Out, In>。
参数循环不再识别任何名为 uplink 的参数。

两个名字，一个入口

Host 方法用 RemoteStream<Out, In> 声明一条流；Client 持有的是 RemoteStreamHandle<Out, In>。两者都从 @deepseek-ai/dsh-typert-protocol 主入口导出，句柄接口没有任何 Host 依赖。生成的 Client 契约把返回类型写成 RemoteStreamHandle<Out, In>。一个名字只有一个含义：Client 代码从主入口拿到的 RemoteStream 永远是声明类型，不会与句柄混淆。

网关 Host（`packages/api/gateway/src`）

// types.ts
export interface InvokeRemoteRequest {
  readonly namespace: string
  readonly method: string
  readonly args: Readonly<Record<string, unknown>>
  /** Uplink items; absent is equivalent to an iterator that ends immediately. */
  readonly uplink?: AsyncIterable<unknown>
  /** The Peer that initiated the call; required on every path. */
  readonly peer: PeerScope
  readonly signal?: AbortSignal
}

export interface TypertGatewayWireStream {
  open: (endpoint: string, payload: unknown, uplink: AsyncIterable<unknown>, peer: PeerScope, signal: AbortSignal) => Promise<AsyncIterable<unknown>>
  failure: (error: unknown) => RemoteStreamFailure
}

// stream-server.ts
export type RemoteStreamOpener = (
  endpoint: string,
  payload: unknown,
  uplink: AsyncIterable<unknown>,
  peer: PeerScope,
  control: AbortController,
  window: RemoteUplinkWindow,
) => Promise<AsyncIterable<unknown>>

// index.ts Config addition
/** Uplink inbox limit per logical stream, in UTF-8 bytes of frames (default 262144). */
readonly streamInboxBytes?: number

prepareInvocation：

const args = await Promise.all(descriptor.parameters.map(parameter => this.resolveParameter(parameter, request.args, endpoint)))
const signal = request.signal ?? NEVER_ABORTED_SIGNAL
const invocation = new GatewayInvocation(
  { namespace: request.namespace, method: request.method, args: request.args },
  descriptor.service,
  request.peer,
  signal,
  new UplinkDecoder(request.uplink ?? EMPTY_ASYNC_ITERABLE, descriptor.uplink?.codec, endpoint, control),
)
const callReceiver = receiverContext.extend({ invocation }).get(descriptor.service) as object
if (descriptor.cancellation !== undefined) args.push(signal)

GatewayInvocation.uplink() 第一次调用返回解码器，第二次抛错。UplinkDecoder 是手写迭代器：有 codec 时逐项 decode(codec, value, endpoint, 'uplink')，失败即 control.abort(failure) 并抛出；无 codec 时只做 isRemoteJsonValue 校验。cancellableStream 的 finally 先关闭解码器（invocation.close()）再等待方法迭代器的 return()，阻塞在 uplink.next() 上的方法得以退出。unary 与 stream 的模式分支恢复为今天的两分支。SRC 回退：描述符无 uplink codec，uplink() 交付 unknown。

网关 Host mux（`stream-server.ts`）

interface ActiveStream { readonly control: AbortController; readonly inbox: UplinkInbox; done: Promise<void> }

/** Bounded uplink queue; iterated as the source of uplink(). */
class UplinkInbox implements AsyncIterable<unknown>, AsyncIterator<unknown> {
  constructor(maxBytes: number, onOverflow: (error: RemoteError) => void)
  push(value: unknown, frameBytes: number): void   // push after ended fails the stream with gateway/protocol; overflow calls onOverflow
  end(): void                                        // idempotent
  fail(error: unknown): void                         // ends the iterator with an error on cancel or carrier closure
  next() / return()                                  // hand-written: next races failure, return releases immediately
}

receive(text) 分派：

open   : exists → throw (socket 1008); otherwise create { control, inbox, done } and start the pump
item   : missing → ignore; otherwise inbox.push(value, bytes)
end    : missing → ignore; otherwise inbox.end()
cancel : missing → ignore; otherwise control.abort(new Error('Remote stream cancelled'))

pump 把 inbox 与该 socket 的 Peer 传给 opener；finally 里 inbox.fail(...)。以 RemoteError 为 reason 的中止由 pump 作为 error 帧发出（网关或 mux 发起的失败）；以普通 Error 为 reason 的中止（cancel、socket 关闭）不发终止帧。

网关 Client（`packages/api/gateway/src/client`）

prepareInvocation 不再剥离任何上行参数；位置参数仍是业务参数加可选 signal。
invokeStream 返回 RemoteStreamHandle：内部持有一个上行队列，send 入队并由泵发 item 帧，end 发 end 帧，dispose 中止代际 signal。下行仍是 openRemoteStream 返回的迭代器。
RemoteStreamMuxClient.open(endpoint, payload, signal, uplink?)：在发出 open 帧后启动上行泵，逐项发 item，源结束发 end；下行终止或 signal 中止时停泵并调用源的 return()（不等待，避免键盘类生成器卡死）；源抛错时 inbox.fail(error)，下行以该错误失败，finally 按既有规则发 cancel。泵检查 this.socket === socket，旧代际不往新 socket 上发帧。
进程内载体：connection.rpc.open(channel, endpoint, payload, signal, uplink)。

落点

包	承载
`dsh-typert-protocol`（`types.ts`、`index.ts`）	`RemoteStream`、`RemoteStreamHandle`、`PeerId`、`PeerScope`、`RemoteInvocation`、`ctx.invocation` 声明合并、`InvocationDescriptor.uplink`；修饰器只认 `mode: 'stream'`；`TypertRemoteService` 构造时注册 `invocation` accessor
`dsh-typert-registry`	加载校验 `uplink.codec` 为有效的严格 codec
`dsh-typert-generator`（`model.ts`、`analyzer.ts`、`emitter.ts`）	`remoteResultType` 识别 `RemoteStream<Out, In>` 并生成 `uplink` boundary；描述符输出 `uplink: { codec }`；生成的 Client 签名返回 `RemoteStreamHandle<Out, In>`；参数循环只认业务参数与末位 `signal`
`dsh-client-connection`（`peer.ts`、`rpc.ts`、`rpc-host.ts`、`index.ts`）	`Peer` 类；`connection.browserPeer` 与 `admit`；`/api` 与升级路由经 `admit` 取 Peer；`rpc.open` 的可选 `uplink`
`dsh-api-gateway` Host（`stream-protocol.ts`、`stream-server.ts`、`types.ts`、`index.ts`）	`item` / `end` 帧；`UplinkInbox`；`Config.streamInboxBytes`；`GatewayInvocation` 与 `extend({ invocation })`；`UplinkDecoder`；`handleUpgrade(req, socket, head, peer)`
`dsh-api-gateway` Client（`client/index.ts`、`client/stream-client.ts`）	`invokeStream` 返回句柄；上行泵由句柄的队列驱动
`dsh-webworker-runtime`、`dsh-remote-mock`	worker 隧道的两种上行帧与 `serveStream` 组装；`StreamHandle.uplink`；直接代理返回真句柄
文档	gateway、api 组、connection、protocol、generator 的 README；`docs/api-gateway`；`docs/subsystems/typert` 的 type-equiv 块；config 与 Cordis 目录

关键流程时序

打开、上行、半关闭、下行结束

Client caller      mux client              mux server              gateway                 Host method
   │ attach(req)        │                       │                     │                         │
   │───────────────────▶│ open { id, ep, payload } ──────────────────▶│ create inbox            │
   │ s.send(v1)         │ item { id, v1 } ───────────────────────────▶│ inbox.push(v1)          │
   │                    │                       │                     │ prepareInvocation       │
   │                    │                       │                     │ extend({ invocation })  │
   │                    │                       │                     │── method(req, signal) ─▶│
   │                    │                       │                     │                         │ uplink() → v1
   │ s.send(v2)         │ item { id, v2 } ───────────────────────────▶│ inbox.push(v2)          │ → v2
   │ s.end()            │ end { id } ────────────────────────────────▶│ inbox.end()             │ iteration ends
   │◀── yield o1 ───────│◀────────────────────── item { id, o1 } ─────│◀── yield o1 ────────────│
   │◀── ends ───────────│◀────────────────────── end { id } ──────────│◀── method returns ──────│ gateway calls uplink.return()

客户端 dispose

Client caller      mux client              mux server              gateway                 Host method
   │ s.dispose()        │                       │                     │                         │
   │───────────────────▶│ stop pump; source return() (not awaited)    │                         │
   │◀── iterator ends ──│ cancel { id } ─────────────────────────────▶│ control.abort()         │
   │                    │                       │                     │ cancellableStream: invocation.close() closes uplink
   │                    │                       │                     │ iterator.return() ─────▶│ finally cleanup
   │                    │                       │ no terminal frame   │                         │

inbox 超限

Client caller      mux client              mux server                        gateway / Host method
   │ s.send(vN)         │ item { id, vN } ─────▶│ inbox.push: bytes > streamInboxBytes │
   │                    │                       │ onOverflow → control.abort(RemoteError uplink-overflow)
   │                    │                       │                                      │ uplink() ends with that error; method iterator return()
   │◀── throw ──────────│◀──────────────────────│ error { id, uplink-overflow }: the pump sends an error frame because the reason is a RemoteError

边界矩阵

事件	opening	running	上行已 end	下行已终止
收到 `item`	入 inbox 缓冲	入 inbox	流以 `gateway/protocol` 失败	忽略
收到 `end`	inbox.end	inbox.end	幂等	忽略
收到 `cancel`	abort	abort	abort	忽略
inbox 超限	流以 `uplink-overflow` 失败	同	不可能	不可能
解码失败	不可能	流以 `input-invalid` 失败	同	不可能
Host 方法结束	不可能	发 `end`；`uplink.return()`	发 `end`	已发
Host 方法抛错	发 `error`	发 `error`；`uplink.return()`	发 `error`	已发
socket 关闭	abort，等 done	同	同	无事
Client `send`	入队，open 后发	发	抛错	抛错
Client `end`	记录，open 后发	发	幂等	忽略
Client 侧下行先结束	不可能	停泵；源 `return()`；不发 `end`	无事	无事

错误码

错误码	何时
`gateway/input-invalid`	既有：参数 codec 解码失败；新增：上行项解码失败
`gateway/uplink-overflow`	新增：inbox 超过 `streamInboxBytes`
`gateway/protocol`	新增：`end` 之后的 `item`
`gateway/cancelled`	既有：`signal` 中止

新码登记在 packages/api/gateway/src/remote-error-codes.ts 与 README 的错误码一节。

最小 echo 方法

packages/api/gateway/tests/echo-stream.host.spec.ts 让同一个方法分别穿过 WebSocket 载体与进程内载体：

class EchoService extends TypertRemoteService {
  constructor(ctx: Context) { super(ctx, 'echo', { namespace: 'echo' }) }

  @Remote({ mode: 'stream' })
  async *echo(prefix: string, signal: AbortSignal): RemoteStream<string, string> {
    for await (const item of this.ctx.invocation!.uplink<string>()) {
      signal.throwIfAborted()
      yield `${prefix}${item}`
    }
  }
}

// Client
const stream = remote.echo.echo('> ')
stream.send('a'); stream.send('b'); stream.end()
const replies: string[] = []
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' })。 字符串引用类型名不受编译器约束，重命名即断。返回类型别名把两个类型写在同一个编译器可见的位置。

运行时传 schema：uplink(schema)。 不经生成器，客户端失去类型投影，与"每项严格 codec"的体系分裂。codec 的来源仍是生成器，方法里的泛型只是断言。

analyzer 交叉校验 uplink<In>() 手填的泛型与返回类型一致。 业务特化的校验，写错后果由调用方承担，与仓库其他处手填断言一致；不加。

传输层句柄带重开、游标、reopen 钩子。 断线重开是上层协议的事：job 的续传是 opened 帧带 from、客户端记住 next 再调一次；snapshot 与 journal 各有自己的锚点与分类。传输层只认一个代际。

客户端 send 放在生成方法返回的 AsyncIterable 上。 AsyncIterable 上挂方法不常规；句柄类型 RemoteStreamHandle 显式扩展它。

照搬 Access 机制。 访问控制是业务语义，vouch 与 remote/invoke waterfall 为它服务；本 Note 只需要"这条流是谁的"，PeerScope 就够。方法级访问控制后来归入 Peer 身份属性与 Remote 方法级访问控制，该部分由其 supersede。

继续用 unary RPC 做上行。 每次上行一次 HTTP 往返、一次鉴权、一次查找；两条载体没有顺序关系，只能靠客户端串行化；没有 EOF；取消不覆盖两半；每个功能各造一套关联键。

原生 HTTP 请求体流。 一次性请求体，与下行流没有关联，不经 Typert 描述符，浏览器对全双工 fetch 的支持不可依赖。

每条双工流一个独立 WebSocket。 失去与现有下行流共享的鉴权、心跳与多路复用，每个功能要自己管一条 socket 的生命周期。

WebTransport 或 HTTP/2 双向流。 有每流流控，但浏览器与 Node 两侧的支持、代理穿透与 Electron 载体都不成熟；在现有 mux 上加两种帧比换传输层便宜几个量级。

后果

买到的：一条流两个方向，终端按键、审批回答、后台任务 stdin 这类交互式场景不再各造一套"一条流加一个 unary"；Host 方法知道调用者；上行类型与下行类型在返回类型里一处声明，上行项来自浏览器，在 Host 逐项严格校验，下行项是 Host 自产的类型值，直接透传；既有 AsyncIterable<Out> 方法与 unary 方法的生成物与线路帧完全不变。
this.ctx.invocation 在非 Remote 调用下是 undefined，方法读它要处理可选；一个只被进程内直接调用的服务方法读到 undefined 是正确行为。
Web 终端使用 attach 传输屏幕输出和输入命令，并提供提供者确认和附件专属可写状态；Client 保留当前代次的句柄，绝不重放输入。
inbox 上限对所有流生效：一个不读上行的方法收到大量上行帧会整流失败。这是刻意的显式失败，README 记明。
所有浏览器连接共享 browserPeer，admit 是唯一的网络接纳点；进程内代码经 connection.createPeer() 创建其他 Peer。
上行与下行之间没有跨方向顺序保证；需要请求应答配对的协议自带序号。
上层监督器仅暴露下行：$stream、snapshot 和 journal 不暴露 send；需要写入上行的领域保留该代次的生成物理句柄。
worker 隧道与 WebSocket 载体行为不同：没有 inbox 上限，end 之后的项丢弃，因为页面到 worker 是同源可信边界。

测试

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 帧并结束后续读取；泵的三种退出；终止帧到达即停泵。
typert：analyzer 对 RemoteStream<Out, In>、RemoteStream<Out>、never 与既有 AsyncIterable<Out> 的处理；修饰器选项；registry 加载校验。
connection：admit 的 401、403 与 browserPeer 三种结果；browserPeer 经桥与共享处理器到达；随 connection 释放。
remote-mock、worker 隧道：上行透传；直接代理返回真句柄。
既有 stream 与 unary 方法的测试不改期望值即通过。

后续

$stream / snapshot / journal 把句柄的 send 透出给上层协议。
双工 $events：waterfall 的回答成为同一 streamId 上的上行项，$events/result 与 clientId 退场。
credit 帧：等需要持续大流量上行的消费者出现；inbox 上限就是初始授信。
Host 发起的半关闭帧（Host 不再读上行但继续写下行）：没有消费者，不加。
unary 方法的 uplink()：能读到方法运行期间到达的项，语义成立但用处存疑；允许，README 注明。

Remote duplex stream: one kind of stream, an uplink channel, and an invocation context

English

Problem

Current state

Uplink consumers

Consumer responsibilities

Relation to background jobs

Goals

Decision

In one sentence

Terms

Type signatures

Host face

How the invocation context reaches the method

PeerScope

Client face

Wire frames

Logical stream state machine

Host side

Client side

Half-close and termination

Cancellation wait lifetime

Ordering guarantees

Backpressure

Security and validation

In-process and worker carriers

Relation to $events

Type and wire definitions

typert protocol (packages/typert/protocol/src)

typert generator (packages/typert/generator/src)

Two names, one entry point

Gateway Host (packages/api/gateway/src)

Gateway Host mux (stream-server.ts)

Gateway Client (packages/api/gateway/src/client)

Placement

Key sequences

Open, uplink, half-close, downlink end

Client dispose

inbox overflow

Boundary matrix

Error codes

Minimal echo method

Alternatives considered

Consequences

Testing

Deferred

Related

中文

问题

现状

上行消费者

消费者职责

与后台 job 的关系

目标

决定

一句话

术语

类型签名

Host 面

调用上下文如何到达方法

PeerScope

Client 面

线路帧

逻辑流状态机

Host 侧

Client 侧

半关闭与结束语义

取消等待的生命周期

顺序保证

背压

安全与校验

进程内与 worker 载体

与 $events 的关系

类型与线路定义

typert protocol（packages/typert/protocol/src）

typert generator（packages/typert/generator/src）

两个名字，一个入口

网关 Host（packages/api/gateway/src）

网关 Host mux（stream-server.ts）

网关 Client（packages/api/gateway/src/client）

Relation to `$events`

typert protocol (`packages/typert/protocol/src`)

typert generator (`packages/typert/generator/src`)

Gateway Host (`packages/api/gateway/src`)

Gateway Host mux (`stream-server.ts`)

Gateway Client (`packages/api/gateway/src/client`)

与 `$events` 的关系

typert protocol（`packages/typert/protocol/src`）

typert generator（`packages/typert/generator/src`）

网关 Host（`packages/api/gateway/src`）

网关 Host mux（`stream-server.ts`）

网关 Client（`packages/api/gateway/src/client`）