feat(client): duplex stream channels with a send path (#107)

* feat(client-core): let a subscription speak first, and again after a reconnect

subscribe() takes hello and goodbye frames. hello goes out once the
transport reports open and after every reconnect, in subscription
order; goodbye goes out on release while the socket is up. A transport
without send() refuses a hello at subscribe time rather than dropping
it. The snapshot says which channels carry a hello, and attempts may be
Infinity, which the reconnect loop already honoured.

* fix(client-core): send hello once, order onReconnect after greet, guard goodbye-only

Three defects from review. First, a first subscribe on a transport
with no onOpen greeted the new hello inside open() and then sent it
again in subscribe() itself; subscribe() now skips the explicit send
when this call is the one that opened the socket. Second, onReconnect
fired right after open() returned even on a transport whose open event
has not happened yet, so a consumer refetching on reconnect could run
before the hello reintroducing the subscription went out; open() now
takes the reconnect's channel list and reports it from the ready path,
after greet(), never before. Third, the no-send guard only checked
hello, so a goodbye-only subscription on a connection without send()
returned normally and later dropped its goodbye in silence; the guard
now applies whenever either option is present.

Updated the gap-recovery and reconnect tests that relied on the old,
premature onReconnect timing to call open() on the reconnected fake
connection, matching what a real transport requires before the manager
can call it ready.

* fix(client-core): carry hello/goodbye through a repartition reopen too

repartition() had the same two defects reconnect() just had fixed.
It called onReconnect right after open(), before the replacement
socket's transport had actually reported open, so a consumer acting
on onReconnect could run before the replacement was greeted. And it
copied channels and refs onto the replacement socket but never
frames, so every surviving subscription's hello and goodbye
registration was silently dropped on an identity change: no hello
went out on the reopened socket and the server never learned the
subscription existed again.

frames now carries over in the same insertion order, and repartition
routes its onReconnect through open()'s report parameter, so it fires
from the ready path after greet(), exactly like a drop-triggered
reconnect.

* feat(client-core): a duplex binding kind the binder hands over raw

StreamBinding is now a union: the entity binding everyone generates
today, with an optional kind, and a duplex binding for a channel the
client speaks on with no entity behind it. The binder keeps duplex
channels out of the entity store and exposes raw(channel, handler,
options), which subscribes through the manager with hello and goodbye
frames and refuses any channel the generated table does not declare.

Three existing tests read StreamBinding fields the brief did not list:
frame-ordering.test.ts and overlay.test.ts built entity bindings typed
as the bare union and passed them to a frame() helper typed the same
way, and live.test.ts read .message off a ChannelBindings.bindings
array. All three are narrowed to EntityStreamBinding, the first two by
retyping the fixtures and the helper, the last with an isDuplex filter,
rather than cast.

stream binding measured 4040 B against its 3.75 kB budget once raw()
and isDuplex landed. Raised to 4.2 kB, which clears the measurement
with about the same headroom the query engine's own budget carries.

* feat(client-gen): emit a duplex binding for a send-and-receive channel

A WebSocket channel that declares both a send and a receive operation
and no entity is one the client speaks on; the binder takes its frames
raw. writeStreams now emits it with kind 'duplex' and the two message
names, and stamps kind 'entity' on the bindings it already emitted.
The kind is optional on entity bindings in client-core, so every table
generated before this keeps type-checking.

* fix(client-devtools): render duplex stream bindings in the streams tab

The bindings table read message, entity, intent and invalidates off every
binding. With StreamBinding now a union that no longer compiles, and at
runtime the first duplex binding threw and blanked the whole tab. The table
narrows on kind: an entity row keeps its columns and a duplex row shows the
send and receive message names with the entity columns empty.

The socket snapshot rows now carry hello, so the live-panel test expects
it, and the streams test narrows its binding assertion for the union.

Author: juicycleff

internal/client/generators/typescript/e2e_streamonly_manifest_test.go

@@ -145,3 +145,68 @@ func TestSpecWithNoSurfaceGetsNoManifest(t *testing.T) {
 		t.Error("ops.ts was emitted for a spec with neither endpoints nor channels")
 	}
 }
+
+// A channel that declares a send and a receive operation and no entity is a
+// duplex channel: the client speaks on it and takes the frames raw. The
+// generator derives that from the AsyncAPI alone, so a backend that already
+// declares both operations needs no extension key to reach the client.
+func TestDuplexChannelGetsADuplexBinding(t *testing.T) {
+	spec := streamOnlySpec()
+	spec.WebSockets = append(spec.WebSockets, client.WebSocketEndpoint{
+		ID:            "liveQueryWS",
+		Path:          "/api/v1/query/live/ws",
+		SendSchema:    &client.Schema{Type: "object"},
+		ReceiveSchema: &client.Schema{Type: "object"},
+		Metadata: map[string]any{
+			"messages": map[string]string{"send": "send", "receive": "receive"},
+		},
+	})
+
+	config := baseConfig()
+	config.Hooks = true
+	config.IncludeStreaming = true
+
+	out, err := NewGenerator().Generate(context.Background(), spec, config)
+	if err != nil {
+		t.Fatalf("Generate: %v", err)
+	}
+
+	ops := ClientManifestText(out.Files)
+
+	for _, want := range []string{
+		"kind: 'duplex'",
+		"channel: '/api/v1/query/live/ws'",
+		"send: 'send'",
+		"receive: 'receive'",
+		"kind: 'entity'",
+	} {
+		if !strings.Contains(ops, want) {
+			t.Errorf("stream bindings are missing %q in:\n%s", want, ops)
+		}
+	}
+}
+
+// A receive-only channel without an entity is neither: it stays out of the
+// table exactly as before, so a listen-only endpoint does not become a
+// subscribable channel by accident.
+func TestReceiveOnlyChannelWithoutEntityStaysOut(t *testing.T) {
+	spec := streamOnlySpec()
+	spec.WebSockets = append(spec.WebSockets, client.WebSocketEndpoint{
+		ID:            "ticker",
+		Path:          "/api/v1/ticker",
+		ReceiveSchema: &client.Schema{Type: "object"},
+		Metadata:      map[string]any{"messages": map[string]string{"tick": "receive"}},
+	})
+
+	config := baseConfig()
+	config.IncludeStreaming = true
+
+	out, err := NewGenerator().Generate(context.Background(), spec, config)
+	if err != nil {
+		t.Fatalf("Generate: %v", err)
+	}
+
+	if strings.Contains(ClientManifestText(out.Files), "/api/v1/ticker") {
+		t.Errorf("a receive-only channel without an entity must not be emitted")
+	}
+}

internal/client/generators/typescript/opsmanifest.go

@@ -680,12 +680,30 @@ func (g *OpsManifestGenerator) writeStreams(buf *strings.Builder, spec *client.A
 		bindings []client.StreamBinding
 	}
 
+	type duplex struct {
+		path    string
+		send    string
+		receive string
+	}
+
 	channels := make([]channel, 0, len(spec.WebSockets)+len(spec.SSEs))
+	duplexes := make([]duplex, 0)
 
 	for i := range spec.WebSockets {
-		if b := spec.WebSockets[i].StreamBindings; len(b) > 0 {
-			channels = append(channels, channel{spec.WebSockets[i].Path, b})
+		ws := &spec.WebSockets[i]
+
+		if b := ws.StreamBindings; len(b) > 0 {
+			channels = append(channels, channel{ws.Path, b})
+
+			continue
 		}
+
+		if ws.SendSchema == nil || ws.ReceiveSchema == nil {
+			continue
+		}
+
+		send, receive := duplexMessageNames(ws.Metadata)
+		duplexes = append(duplexes, duplex{ws.Path, send, receive})
 	}
 
 	for i := range spec.SSEs {
@@ -695,12 +713,14 @@ func (g *OpsManifestGenerator) writeStreams(buf *strings.Builder, spec *client.A
 	}
 
 	sort.Slice(channels, func(i, j int) bool { return channels[i].path < channels[j].path })
+	sort.Slice(duplexes, func(i, j int) bool { return duplexes[i].path < duplexes[j].path })
 
 	buf.WriteString("export const streams = [\n")
 
 	for _, ch := range channels {
 		for _, b := range ch.bindings {
 			buf.WriteString("  {\n")
+			buf.WriteString("    kind: 'entity',\n")
 			buf.WriteString(fmt.Sprintf("    channel: %s,\n", tsString(ch.path)))
 			buf.WriteString(fmt.Sprintf("    message: %s,\n", tsString(b.Message)))
 			buf.WriteString(fmt.Sprintf("    entity: %s,\n", tsString(b.EntityType)))
@@ -710,9 +730,44 @@ func (g *OpsManifestGenerator) writeStreams(buf *strings.Builder, spec *client.A
 		}
 	}
 
+	// A duplex channel: the client speaks on it and no entity stands behind it.
+	// Derived from the declared operations alone, so the backend needs no
+	// extension key for a channel it already describes as send and receive.
+	for _, d := range duplexes {
+		buf.WriteString("  {\n")
+		buf.WriteString("    kind: 'duplex',\n")
+		buf.WriteString(fmt.Sprintf("    channel: %s,\n", tsString(d.path)))
+		buf.WriteString(fmt.Sprintf("    send: %s,\n", tsString(d.send)))
+		buf.WriteString(fmt.Sprintf("    receive: %s,\n", tsString(d.receive)))
+		buf.WriteString("  },\n")
+	}
+
 	buf.WriteString("] as const;\n")
 }
 
+// duplexMessageNames picks the lowest-sorted message name for each direction,
+// the same tie-break convertSchemaFromChannel uses, so the output is stable
+// across runs.
+func duplexMessageNames(metadata map[string]any) (string, string) {
+	names, _ := metadata["messages"].(map[string]string)
+	send, receive := "", ""
+
+	for _, name := range sortedKeys(names) {
+		switch names[name] {
+		case "send":
+			if send == "" {
+				send = name
+			}
+		case "receive":
+			if receive == "" {
+				receive = name
+			}
+		}
+	}
+
+	return send, receive
+}
+
 // tsString renders a single-quoted TypeScript string literal.
 //
 // Escaping is not paranoia: a path or tag reaching the file unescaped closes the

packages/client-core/__tests__/duplex-binding.test.ts

@@ -0,0 +1,80 @@
+import { describe, expect, it } from 'vitest';
+
+import { QueryCache } from '../src/cache';
+import { manualScheduler } from '../src/invalidate';
+import { binderSnapshot, StreamBinder } from '../src/live';
+import { SubscriptionManager } from '../src/stream';
+import type { StreamBinding } from '../src/stream';
+import { manualClock } from '../src/transport';
+import { fakeSockets, fakeTransport } from './harness';
+
+const streams: readonly StreamBinding[] = [
+  { kind: 'duplex', channel: '/api/v1/query/live/ws', send: 'SendMessage', receive: 'ReceiveMessage' },
+  { channel: '/ws/orders', message: 'orderUpdated', entity: 'Order', intent: 'upsert', invalidates: [] },
+];
+
+function build() {
+  const sockets = fakeSockets();
+  const clock = manualClock();
+  const cache = new QueryCache({ transport: fakeTransport(() => []), entities: { Order: { idField: 'id' } } });
+  const manager = new SubscriptionManager({
+    connect: sockets.connect,
+    sleep: clock.sleep,
+    random: () => 0,
+    release: manualScheduler().schedule,
+  });
+  const binder = new StreamBinder({ cache, streams, manager, scheduler: (flush) => flush(), sleep: clock.sleep });
+
+  return { binder, sockets, manager };
+}
+
+describe('a duplex channel', () => {
+  // Guards that a raw subscription is the manager's subscription: same socket,
+  // same hello, same release.
+  it('subscribes raw with a hello and hands frames over undecoded', () => {
+    const { binder, sockets } = build();
+    const seen: unknown[] = [];
+
+    const release = binder.raw('/api/v1/query/live/ws', (message) => seen.push(message), {
+      hello: { action: 'subscribe', data: { id: 'q1' } },
+      goodbye: { action: 'unsubscribe', data: { id: 'q1' } },
+    });
+    sockets.last().open();
+    sockets.last().deliver({ type: 'snapshot', subscriptionId: 'q1', payload: { rows: [] } });
+
+    expect(sockets.last().sent).toEqual([{ action: 'subscribe', data: { id: 'q1' } }]);
+    expect(seen).toEqual([{ type: 'snapshot', subscriptionId: 'q1', payload: { rows: [] } }]);
+
+    release();
+    expect(sockets.last().sent).toEqual([
+      { action: 'subscribe', data: { id: 'q1' } },
+      { action: 'unsubscribe', data: { id: 'q1' } },
+    ]);
+  });
+
+  // Guards the generated table as the single source of channel names.
+  it('refuses a channel that is not a duplex binding', () => {
+    const { binder } = build();
+
+    expect(() => binder.raw('/ws/orders', () => {})).toThrow(/not a duplex channel/);
+    expect(() => binder.raw('/ws/nowhere', () => {})).toThrow(/not a duplex channel/);
+  });
+
+  // Guards the entity path from a frame it must never see.
+  it('keeps duplex frames out of the entity store', () => {
+    const { binder, sockets } = build();
+
+    binder.raw('/api/v1/query/live/ws', () => {}, { hello: { id: 'q1' } });
+    sockets.last().open();
+    sockets.last().deliver({ type: 'orderUpdated', payload: { id: 'o1', total: 3 } });
+
+    expect(binderSnapshot(binder).queued).toBe(0);
+  });
+
+  it('lists the duplex binding in the snapshot', () => {
+    const { binder } = build();
+
+    const channels = binderSnapshot(binder).channels.map((c) => c.channel).sort();
+    expect(channels).toEqual(['/api/v1/query/live/ws', '/ws/orders']);
+  });
+});