feat(transport): CEP-41 open-ended streams

[ContextVM-org]

Implementing ContextVM/contextvm-docs#40

[1amKhush]

@ContextVM-org

Hey! Awesome work on this draft PR. The architecture looks really solid. (The callToolStream free helper is a great design choice for the consumer API and the deferred-based async iterator for the session handles backpressure beautifully)

I did a deep dive into the code and found a couple of bugs and edge cases to look at before we finalize:

1. ping() consumes two progress values

In writer.ts, the ping() method calls this.nextProgress() twice:

public async ping(): Promise<void> {
  if (!this.active) return;
  const nonce = String(this.nextProgress()); // <--- progress bumped here
  await this.publishFrame(
    buildOpenStreamPingFrame({
      progressToken: this.progressToken,
      progress: this.nextProgress(),         // <--- bumped again!
      nonce,
    }),
  );
}

This means every ping skips a progress number. The nonce should probably just reuse the single progress value generated for that frame. (Note: The writer.test.ts test on line 3749 actually encodes this bug by expecting progress: 3 and nonce: "2"!).2. onClose fires even on abort (Double delete)

2. Abort currently runs close cleanup semantics

OpenStreamSession.finish() always calls onClose, even when the stream finished with an error/abort. Local abort() also calls onAbort, so registry cleanup can run through both paths. Remote aborts via processFrame() go through finish(error) and therefore currently call onClose, not onAbort.

This works because Map.delete() is idempotent, but the lifecycle semantics are muddy. It would be cleaner to split successful close cleanup from failed/aborted cleanup.

3. Unbounded buffers on outbound client sessions

In nostr-client-transport.ts, createOutboundOpenStreamSession hardcodes the limits to Number.MAX\_SAFE\_INTEGER:

maxBufferedChunks: Number.MAX_SAFE_INTEGER,
maxBufferedBytes: Number.MAX_SAFE_INTEGER,

Since this session consumes chunks from the server, a buggy or malicious peer can drive unbounded buffering. This should reuse the configured open-stream policy/defaults, or the receiver/registry should expose a create path that applies those defaults automatically.

4. Missing lastChunkIndex validation in the receiver

The spec mentions that if close.lastChunkIndex is present, the receiver must verify it got all chunks up to that index. In session.ts, the processFrame switch statement for close handles the graceful finish, but it currently ignores the lastChunkIndex field on the frame.

5. Monkey-patching the Writer

In nostr-server-transport.ts, the close and abort methods on the OpenStreamWriter are overridden (monkey-patched) via .bind() to flush the pending JSON-RPC response. This totally works, but it's a bit fragile (e.g. if someone subclasses the writer later). We might want to use a callback hook on the writer constructor (like onClose / onAbort) to trigger the response flush instead.

6. Timers (Just a note)

I noticed the idle timeout, probe timeout, and close grace period timers aren't implemented yet in the session/writer (though they are in the policy types). I'm assuming that's just deferred for later since this is a draft, but wanted to point it out just in case!

7. Stale chunkIndex values can wedge the session

bufferChunk() rejects duplicate chunk indexes only if they are still in bufferedChunks. Once a chunk has already been flushed and nextExpectedChunkIndex has advanced, a later duplicate/stale chunk index is accepted into bufferedChunks.

Example: receive and flush chunk 0, then receive another chunk 0 with higher progress. It gets buffered, but flushContiguousChunks() is waiting for chunk 1, so that stale chunk remains buffered. If close arrives later, maybeFinishGracefully() sees bufferedChunks.size > 0 and the session never finishes.

The receiver should reject frame.chunkIndex < nextExpectedChunkIndex.

Overall, incredibly clean implementation. Really looking forward to seeing this land! :)

[1amKhush]

Btw I’d keep 'Monkey-patching writer methods' only as a non-blocking design comment (The close/abort reassignment is brittle. I’d classify this as design risk, not a correctness bug). Lmk your thoughts on this!

[ContextVM-org]

Please review again @1amKhush . Your comments were attached in the last commit

[1amKhush]

Looking great @ContextVM-org ! Looks great, all the issues I flagged are addressed 👍

Just had 2 minor/ 'good to have' things (the current implementation is great as is, so i think these can be ignored)

• The keepalive timers are fully implemented in session.ts, but the e2e test suite doesn't have a test that actually exercises the idle timeout → automatic ping → pong → continued stream flow. There are unit tests for the session, but an e2e test through the full transport would catch wiring issues. This is a "nice to have" though, not a blocker.
• In createOutboundOpenStreamSession, the client-side sendPing/sendPong/sendAbort callbacks each maintain a local let progress = 0 counter. But the server is also sending frames on the same stream with its own progress counter. The spec says progress must be monotonically increasing across the stream, if the client sends a ping at progress=1 and the server had already sent a chunk at progress=5, the client's ping would have a lower progress value than what the receiver already saw.
  (Though imo this probably doesn't matter in practice since the server is the one sending chunks and the client is the one receiving, and the session validates progress per-direction. Just wanted to know your thoughts on this)

Rest the implementation is really solid at this point! All 6 original issues were fixed properly, the keepalive timers are fully wired, and the buffer limits use real policy values.

[ContextVM-org]

Added these test and some more. Let me know your thoughts

[1amKhush]

Added these test and some more. Let me know your thoughts

Looking great!

[1amKhush]

LGTM, good to merge 👍