Add channel-level flow control and fair delivery

[connor4312]

Motivation

A busy terminal, TCP stream, or telemetry channel cannot crowd out other traffic on the same connection. #488 proposes TCP channels with their own flow control, but imo we can introduced a shared mechanism rather than maintain a second set of window and credit rules.

Proposed direction

• Negotiate receive windows and delivery limits in subscribe, in both directions. The response confirms the accepted settings and the host’s receive limits. These belong to each subscription, not shared channel state.
• Allow logical messages to span multiple small JSON-RPC messages. Carry fragments as strings, not another layer of base64; do not split surrogate pairs. Count UTF-16 bytes and account for JSON escaping when enforcing frame limits. Reassemble before normal typed decoding or reducer application.
• Treat the window as a target: starting a message needs available credit, but an already-started message may finish beyond the window. This avoids requiring callers to predict encoded size or repeatedly resize payloads. Keep hard frame/message limits and bounded buffers so this allowance cannot grow without limit.
• Return credit when the bounded consumer releases the message, not merely when JSON arrives. For TCP, that means releasing bytes from the downstream stream buffer. Use the same cumulative credit mechanism for all channel types.
• Schedule fragments fairly across subscriptions and bound the underlying transport queue. Fragmentation and windows alone do not prevent one busy channel from monopolizing delivery. Small, bounded credit/reset/liveness controls must not depend on data credit.
• Include initial snapshots and reconnect replay in bounded delivery, rather than allowing large RPC results to bypass it. Use per-channel recovery cursors so one stalled channel does not prevent other channels from progressing safely.

Effect on TCP

I would keep TCP byte offsets, duplicate suppression, EOF/close/reset, and same-socket replay. Those describe stream correctness. I would move window negotiation and consumed-credit updates out of TCP-specific state/actions into the shared subscription machinery. Arbitrary TCP bytes would still need their existing encoding on a JSON transport.

This is a design proposal, not a requirement to block #488. The main questions are the precise credit-release contract, bounded completion beyond the window, and how bootstrap/recovery delivery fits the existing APIs. Subscription flow control also needs to be clear about its scope: other large RPC results need separate bounds or chunking to provide connection-wide protection.

[connor4312]

Connor's robot driving:

Here is a more concrete sketch of what I have in mind. Names and shapes are illustrative, not a final API.

1. Negotiate delivery per subscription, in both directions

interface ChannelReceiveLimits {
  windowBytes: number;          // Target outstanding logical-payload bytes.
  maximumFrameBytes: number;    // Hard limit on the encoded outer message.
  maximumMessageBytes: number;  // Hard limit on one reassembled logical message.
}

interface SubscriptionFlowControlOptions {
  receive: ChannelReceiveLimits;
}

interface NegotiatedSubscriptionFlowControl {
  clientReceive: ChannelReceiveLimits;
  hostReceive: ChannelReceiveLimits;
}

For example:

// Client -> host
{
  "jsonrpc": "2.0",
  "id": 12,
  "method": "subscribe",
  "params": {
    "channel": "ahp-terminal:/t1",
    "flowControl": {
      "receive": {
        "windowBytes": 262144,
        "maximumFrameBytes": 16384,
        "maximumMessageBytes": 4194304
      }
    }
  }
}

// Host -> client
{
  "jsonrpc": "2.0",
  "id": 12,
  "result": {
    "subscriptionId": "sub-7",
    "flowControl": {
      "clientReceive": {
        "windowBytes": 262144,
        "maximumFrameBytes": 16384,
        "maximumMessageBytes": 4194304
      },
      "hostReceive": {
        "windowBytes": 131072,
        "maximumFrameBytes": 16384,
        "maximumMessageBytes": 4194304
      }
    },
    "initial": { "kind": "snapshot" }
  }
}

I would capability-gate this mode and require explicit acceptance. The host can lower the requested receive limits, not raise them. Existing non-windowed subscriptions remain unchanged.

subscriptionId identifies this delivery generation, not the channel resource. I would keep it so delayed credit from an old subscription cannot release credit for a new subscription to the same channel. Reconnect establishes a new delivery generation; durable recovery still uses the channel and its replay cursor.

2. Deliver string fragments without needing the total size upfront

interface ChannelFrameParams {
  channel: URI;
  subscriptionId: string;
  messageId: number;
  data: string;
  final?: boolean;
}

interface ChannelCreditParams {
  channel: URI;
  subscriptionId: string;
  consumedBytes: number; // Cumulative, separately for each direction.
}

{
  "jsonrpc": "2.0",
  "method": "channel/frame",
  "params": {
    "channel": "ahp-terminal:/t1",
    "subscriptionId": "sub-7",
    "messageId": 23,
    "data": "{\"jsonrpc\":\"2.0\",\"method\":\"action\","
  }
}

The payload is serialized JSON for an existing typed message. Bootstrap would have typed snapshot/ready messages too. The receiver concatenates fragments and only then decodes and delivers the complete message. Reducers never see fragments.

Only final: true completes the message; omission or false means more fragments follow. Fragment offsets are unnecessary on the reliable, ordered transport. On disconnect, discard incomplete messages and replay whole logical messages using the per-channel recovery cursor rather than resuming partial fragments.

I would allow one active logical message per subscription/direction, with frames interleaved across subscriptions. There is no required total-size field. Fragments must not split actual surrogate pairs. Credit counts UTF-16 bytes of data before outer JSON escaping; the frame cap includes the actual outer serialization and escaping. SDKs can build fitting fragments in one pass rather than repeatedly serializing or binary-searching application payloads.

3. A soft window, with permission to finish the current message

The sender maintains:

const outstandingBytes = sentBytes - consumedBytes;
const canStartMessage =
  !activeMessage && outstandingBytes < receiveLimits.windowBytes;

Once a message starts, the sender may finish it even after using the remaining credit. It must wait for credit before starting another data message. Bytes are reserved before enqueueing each frame, so asynchronous writes cannot reuse the same credit.

For example, with a 256 KiB window and 240 KiB outstanding, a message that turns out to encode to 40 KiB can finish. Outstanding data becomes 280 KiB, then the sender waits.

This avoids requiring callers to predict encoded size. The hard message cap still applies as encoding proceeds, so outstanding payload stays below windowBytes + maximumMessageBytes. Receivers must budget for that allowance, decoding overhead, and aggregate usage across subscriptions. Exceeding a hard limit requires an explicit subscription failure/recovery path, not silent truncation or dropped fragments.

4. One consumption-credit mechanism

{
  "jsonrpc": "2.0",
  "method": "channel/credit",
  "params": {
    "channel": "ahp-terminal:/t1",
    "subscriptionId": "sub-7",
    "consumedBytes": 65536
  }
}

Credit advances through complete messages whose bounded consumer has released them, in order. Applying a state action normally releases it; a TCP adapter releases it when its downstream byte buffer releases the payload. Parsing JSON or receiving an action echo is not sufficient.

Cumulative updates make duplicate/older credit harmless. Credit beyond sent data or between complete-message boundaries is invalid. A new subscription identifier prevents stale credit from affecting a replacement subscription. These counters are delivery bookkeeping, not durable replay cursors.

The sender must schedule frames fairly across subscriptions and bound its transport queue. Small credit/reset/liveness controls bypass data credit, with separate size/rate limits; ordering-sensitive controls still cannot overtake preceding data.

5. TCP uses this instead of its own window actions

Creation would use the same settings:

{
  "channel": "ahp-session:/s1",
  "create": {
    "type": "tcpConnection",
    "host": "localhost",
    "port": 3000,
    "encoding": "base64"
  },
  "flowControl": {
    "receive": {
      "windowBytes": 262144,
      "maximumFrameBytes": 16384,
      "maximumMessageBytes": 4194304
    }
  }
}

The negotiated subscription belongs to the new TCP child, not the parent session.

I would keep tcp/input, tcp/data, byte offsets, EOF/close/reset, duplicate suppression, and same-socket replay. I would remove TCP-specific window negotiation and tcp/inputConsumed / tcp/dataConsumed actions in favor of channel/credit. Adapters stop reading sources when they cannot start another message and return credit when destination buffers release data. Arbitrary TCP bytes still need base64 on a JSON carrier, but framing does not add a second base64 layer. Payload-bearing action echoes also use bounded delivery.

Finally, I would send initial snapshots and reconnect replay through this delivery path after a small acknowledgment, with an explicit ready boundary before live updates. Per-channel replay cursors track safe recovery independently of connection-local credit. For TCP, recovery remains complete replay to the same retained consumer or explicit failure, never restoration from a snapshot.

[kycutler]

From my agent:

---

This looks like the right shared mechanism. Working through how TCP would migrate onto it, alongside terminal and telemetry, I think the core sketch holds up well. I would keep the proposed framing, soft window, consumption-based credit, and per-subscription scheduling.

A few additions and clarifications seem important, particularly for preserving TCP’s existing guarantees:

1. Reconnect must preserve outstanding consumption obligations

Discarding incomplete fragments is straightforward, but complete messages already accepted by a retained consumer must remain charged after reconnect. A new subscriptionId must not grant a fresh window over unread buffers.

There is also a lost-credit case: the host consumes TCP input, its credit update is lost, and the client reconnects. An action replay cursor establishes which actions were delivered, but does not establish which input buffers were subsequently released.

One candidate is a retained per-direction delivery journal, with reconnect exchanging:

• The last accepted message ID and cumulative logical-byte position.
• The last released message ID and cumulative logical-byte position.

The transport-bound subscriptionId would still rotate to fence stale traffic, while message IDs and accounting remain stable within the retained journal. An opaque resume token could identify that journal across reconnect attempts, including a lost reconnect reply.

This is an extension to the sketch, not a replacement for per-channel recovery cursors. The requirement is to preserve retained charge and reconcile lost release updates; the exact representation is open.

2. Shared release receipts need to preserve TCP drain()

Removing tcp/inputConsumed / tcp/dataConsumed should not remove the distinction between action acceptance and downstream consumption.

A TCP adapter should be able to await release of a particular sent message through the shared delivery layer, including when release is learned through reconnect rather than a live credit notification. This avoids retaining a second, TCP-specific consumption mechanism.

Release is still not an application-success acknowledgment: rejected actions release their allocations too, and ordinary action reconciliation must report rejection.

TCP byte offsets remain decoded-byte offsets; shared delivery credit measures serialized logical-message bytes. Those counters serve different purposes.

3. Lifecycle controls must make progress without reordering the stream

EOF/close need a bounded, credit-independent path, but must not overtake preceding data. They can be delivered after that data has been fully delivered without waiting for its consumption; graceful teardown must still honor accepted data.

Reset is different: it can abort incomplete delivery and invalidate the stream. It cannot simply skip a fragment and continue.

Payload-bearing action echoes also need the bounded data path, or they become a backpressure bypass.

4. Bootstrap needs an explicit consumer/readiness contract

The small acknowledgment should install the binding and make a consumer available before snapshot/replay starts. Consumers must run before the ready boundary; otherwise a snapshot larger than the window can deadlock an API that waits for readiness before exposing its reader.

TCP recovery remains complete replay to the same retained socket/consumer, or explicit failure—never snapshot replacement or silently opening another socket.

5. Shared delivery does not imply identical recovery or loss policy

• Terminal: independent viewer pressure; a slow viewer must not stall every viewer or the shared PTY. Snapshot recovery replaces state, and uncertain input must not be blindly retried.
• Telemetry: retain live-only recovery. Bounded overflow/loss policy is separate from credit, and dropping must be explicit—not silent loss of admitted fragments.
• TCP: no lossy continuation.

The main additional protocol question, then, is how retained acceptance/release accounting survives reconnect. Making that explicit should let TCP use the common mechanism without weakening its guarantees or keeping private flow-control machinery alongside it.