fix(runtime): implement the buffered-send streaming fallback on both dispatcher write-less-transport branches

.changeset/dispatcher-streaming-fallback-buffered-send.md

@@ -0,0 +1,5 @@
+---
+"@objectstack/runtime": patch
+---
+
+The dispatcher's two write-less-transport fallbacks for streamed results now implement the IHttpResponse streaming contract's own prescription (#3607, ADR-0076 OQ#10): when a transport's response object lacks the optional `write`/`end` streaming surface, the SSE frames are buffered and delivered through `send()` under the streaming headers, byte-identical to what a streaming transport would have written. Previously the route-wrapper fallback answered a bare JSON `{ events }` body no SSE reader could decode, and the dispatch-result writer fell through to serializing the stream descriptor itself — collapsing its `events` AsyncIterable to `{}` and losing every event silently under HTTP 200. Both branches are reachable only through an externally supplied `Runtime({ server })` transport without streaming support; callers of such compositions that parse the streamed and buffered bodies with the same SSE reader now decode identical frames from either.

packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts

@@ -0,0 +1,305 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * #9936 / #9961 — the write-less-transport fallback for streamed dispatcher
+ * results implements the IHttpResponse contract's own prescription (#3607,
+ * ADR-0076 OQ#10; the JSDoc on `write` in
+ * packages/spec/src/contracts/http-server.ts): buffer the SSE frames and
+ * deliver them through `send()` under the streaming headers.
+ *
+ * ## Why these pins are the entire evidentiary basis
+ *
+ * Both shipped `http.server` providers (plugin-hono-server, the QA node
+ * conformance server) construct `res.write`/`res.end` unconditionally, so
+ * NEITHER fallback branch is reachable in any in-repo composition — they are
+ * live only for an external `Runtime({ server })` transport that omits the
+ * contract's optional streaming surface. No composed run and no gate number
+ * will ever exercise them; these pins are the only evidence they behave.
+ *
+ * ## The property pinned — byte identity, not absence-of-json
+ *
+ * The property that makes the buffered fallback CORRECT is that its body is
+ * byte-identical to the concatenation a streaming transport would have
+ * written: every reader in the measured population (the client SDK's
+ * `parseEventStream`, objectui's @ai-sdk/react DefaultChatTransport) parses
+ * raw response bytes for `data:` lines, so identical bytes parse identically,
+ * streamed or buffered. Each pin below therefore drives the SAME route twice
+ * — once against a streaming `res`, once against a write-less one — and
+ * asserts `sent === writes.join('')`. A pin that only asserted `res.json` is
+ * absent would stay green if the replacement emitted the wrong bytes.
+ *
+ * ## The two sites, and what each pin would catch
+ *
+ * 1. `mountRouteOnServer`'s streaming branch (#9936): used to answer
+ * `res.json({ events })` — an off-envelope JSON dialect of the frames that
+ * decodes to ZERO frames in every SSE reader (a JSON body contains no raw
+ * newline byte). The pin trips on any return to a JSON body (the `json`
+ * spy) and on any byte divergence from the streamed branch.
+ * 2. `sendResultBase`'s dispatch-result writer (#9961): used to fall through
+ * to `res.json(result.result)`, serializing the stream DESCRIPTOR with its
+ * `events` AsyncIterable collapsed to `{}` — HTTP 200, payload gone,
+ * iterable never drained. That body is a RELAYED `res.json(...)`,
+ * invisible to check-route-envelope's counters by design, so no gate
+ * number moves for this half: this pin is the only tripwire. It catches a
+ * regression to descriptor-serialization (json fires, send never does, the
+ * drained flag stays false) and any byte divergence from the streamed
+ * branch, error-frame and `null`-skip encoding included.
+ */
+
+import { describe, it, expect, vi } from 'vitest';
+
+import { createDispatcherPlugin } from './dispatcher-plugin.js';
+
+// ---------------------------------------------------------------------------
+// Harness (shape shared with dispatcher-plugin.route-auth-deny-body.test.ts)
+// ---------------------------------------------------------------------------
+
+function makeFakeServer() {
+ const handlers: Record<string, (req: any, res: any) => any> = {};
+ const rec = (verb: string) => (path: string, handler: any) => {
+ handlers[`${verb} ${path}`] = handler;
+ };
+ return {
+ handlers,
+ server: {
+ get: rec('GET'),
+ post: rec('POST'),
+ put: rec('PUT'),
+ delete: rec('DELETE'),
+ patch: rec('PATCH'),
+ },
+ };
+}
+
+function makeCtx(fakeServer: any, aiRoutes: Array<Record<string, unknown>>) {
+ // The dispatch path only reads the AI route table once an `ai` service
+ // resolves (a service-less kernel answers the 501 capability envelope
+ // before any route is consulted), so the kernel registers a bare one —
+ // the same shape domain-handler-registry.test.ts drives dispatch with.
+ const services: Record<string, any> = { ai: { name: 'ai' } };
+ const kernel: any = {
+ getState: () => 'running',
+ getService: (name: string) => services[name] ?? null,
+ getServiceAsync: async (name: string) => services[name] ?? null,
+ context: { getService: (name: string) => services[name] ?? null },
+ // The AIServicePlugin's cross-plugin cache the dispatcher recovers
+ // routes from when the `ai:routes` hook fired before it was listening.
+ __aiRoutes: aiRoutes,
+ };
+ return {
+ getKernel: () => kernel,
+ getService: (name: string) => (name === 'http.server' ? fakeServer.server : undefined),
+ environmentId: undefined,
+ logger: { info() {}, warn() {}, error() {}, debug() {} },
+ hook: () => {},
+ on: () => {},
+ } as any;
+}
+
+/**
+ * A recording response. `withWriteEnd: true` models a streaming transport;
+ * `false` models the write-less transport both fallbacks exist for — `write`
+ * and `end` are then ABSENT (not stubbed), exactly how the sites
+ * feature-detect them. `settled` resolves on the first terminal write
+ * (`json`, `send`, or `end`) so a regression fails fast instead of timing
+ * out, and `terminal` names which one fired.
+ */
+function recordingRes(withWriteEnd: boolean) {
+ const rec: {
+ status?: number;
+ headers: Record<string, string>;
+ writes: string[];
+ sent?: string;
+ jsonBody?: unknown;
+ terminal?: 'json' | 'send' | 'end';
+ } = { headers: {}, writes: [] };
+ let resolveSettled!: () => void;
+ const settled = new Promise<void>((resolve) => { resolveSettled = resolve; });
+ const terminal = (kind: 'json' | 'send' | 'end') => {
+ if (!rec.terminal) rec.terminal = kind;
+ resolveSettled();
+ };
+ const res: any = {
+ status(code: number) { rec.status = code; return res; },
+ header(k: string, v: string) { rec.headers[k] = v; return res; },
+ json(body: unknown) { rec.jsonBody = body; terminal('json'); return res; },
+ send(data: string) { rec.sent = data; terminal('send'); return res; },
+ };
+ if (withWriteEnd) {
+ res.write = (chunk: string) => { rec.writes.push(String(chunk)); };
+ res.end = () => { terminal('end'); };
+ }
+ return { rec, res, settled };
+}
+
+/**
+ * The reader population's parse, in miniature: the client SDK's
+ * `parseEventStream` and @ai-sdk/react's transport both split raw bytes on
+ * newlines and decode `data:`-prefixed lines. Identical bytes ⇒ identical
+ * frames; this turns the byte-identity assertion into decoded frames too, so
+ * a wrong-bytes regression is reported in reader terms.
+ */
+function decodeSseFrames(body: string): unknown[] {
+ const frames: unknown[] = [];
+ for (const line of body.split('\n')) {
+ const trimmed = line.trim();
+ if (!trimmed.startsWith('data:')) continue;
+ const payload = trimmed.slice(5).trim();
+ if (!payload) continue;
+ frames.push(JSON.parse(payload));
+ }
+ return frames;
+}
+
+/** One string frame and one object frame — both arms of the encoding ternary. */
+const STRING_FRAME = 'data: {"type":"text-delta","delta":"hi"}\n\n';
+const OBJECT_FRAME = { type: 'finish' };
+const EXPECTED_FRAMES = [{ type: 'text-delta', delta: 'hi' }, { type: 'finish' }];
+
+// ---------------------------------------------------------------------------
+// Site 1 — mountRouteOnServer's streaming branch (#9936)
+// ---------------------------------------------------------------------------
+
+describe('mountRouteOnServer write-less fallback (#9936)', () => {
+ const ROUTE = '/ai/stream-pin';
+ const MOUNTED = `PATCH /api/v1${ROUTE}`;
+
+ async function drive(withWriteEnd: boolean) {
+ const fakeServer = makeFakeServer();
+ const drained = { value: false };
+ const handler = vi.fn(async () => ({
+ status: 200,
+ stream: true,
+ events: (async function* () {
+ yield STRING_FRAME;
+ yield OBJECT_FRAME;
+ drained.value = true;
+ })(),
+ }));
+ // PATCH: a legal RouteDefinition.method that only the concrete
+ // hook-route mounts serve (the /ai/* wildcards cover get/post/delete/
+ // put), so this pins the same arm that is live unshadowed on the wire.
+ const ctx = makeCtx(fakeServer, [{ method: 'PATCH', path: ROUTE, auth: false, handler }]);
+ const plugin = createDispatcherPlugin({ prefix: '/api/v1', securityHeaders: false });
+ await plugin.start?.(ctx);
+
+ const mounted = fakeServer.handlers[MOUNTED];
+ expect(mounted).toBeTypeOf('function');
+ const { rec, res, settled } = recordingRes(withWriteEnd);
+ await mounted({ headers: {}, body: {}, params: {}, query: {} }, res);
+ await settled;
+ return { rec, drained };
+ }
+
+ it('buffers the SAME bytes a streaming transport receives, through send() under text/event-stream', async () => {
+ const streamed = await drive(true);
+ const buffered = await drive(false);
+
+ // The streamed leg is the reference measurement.
+ expect(streamed.rec.terminal).toBe('end');
+ expect(streamed.rec.writes.join('')).toContain('data:');
+
+ // ⭐ The property that makes option B correct: byte identity.
+ expect(buffered.rec.terminal).toBe('send');
+ expect(buffered.rec.sent).toBe(streamed.rec.writes.join(''));
+
+ // And in the reader population's terms: identical decoded frames.
+ expect(decodeSseFrames(buffered.rec.sent!)).toEqual(EXPECTED_FRAMES);
+ expect(decodeSseFrames(streamed.rec.writes.join(''))).toEqual(EXPECTED_FRAMES);
+
+ // The old body — res.json({ events }) — must not come back.
+ expect(buffered.rec.jsonBody).toBeUndefined();
+
+ // Same streaming headers on both legs; the iterable was drained.
+ for (const leg of [streamed, buffered]) {
+ expect(leg.rec.status).toBe(200);
+ expect(leg.rec.headers['Content-Type']).toBe('text/event-stream');
+ expect(leg.drained.value).toBe(true);
+ }
+ });
+});
+
+// ---------------------------------------------------------------------------
+// Site 2 — sendResultBase's dispatch-result writer (#9961)
+// ---------------------------------------------------------------------------
+
+describe('dispatch-result writer write-less fallback (#9961)', () => {
+ const ROUTE_PATH = '/api/v1/ai/stream-pin2';
+ const WILDCARD = 'POST /api/v1/ai/*';
+
+ async function drive(withWriteEnd: boolean, events: () => AsyncIterable<unknown>) {
+ const fakeServer = makeFakeServer();
+ const handler = vi.fn(async () => ({ status: 200, stream: true, events: events() }));
+ const ctx = makeCtx(fakeServer, [{ method: 'POST', path: ROUTE_PATH, auth: false, handler }]);
+ const plugin = createDispatcherPlugin({ prefix: '/api/v1', securityHeaders: false });
+ await plugin.start?.(ctx);
+
+ const wildcard = fakeServer.handlers[WILDCARD];
+ expect(wildcard).toBeTypeOf('function');
+ const { rec, res, settled } = recordingRes(withWriteEnd);
+ await wildcard({ path: ROUTE_PATH, headers: {}, body: {}, params: {}, query: {} }, res);
+ // The drain is detached by design (sendResultBase is synchronous by
+ // signature); the response arrives when the iterable settles.
+ await settled;
+ return rec;
+ }
+
+ it('drains the descriptor and buffers the SAME bytes a streaming transport receives — never the serialized descriptor', async () => {
+ const mk = () => {
+ const drained = { value: false };
+ const events = async function* () {
+ yield STRING_FRAME;
+ yield null; // the streamed branch skips null events; the buffer must too
+ yield OBJECT_FRAME;
+ drained.value = true;
+ };
+ return { events, drained };
+ };
+
+ const a = mk();
+ const streamed = await drive(true, a.events);
+ const b = mk();
+ const buffered = await drive(false, b.events);
+
+ expect(streamed.terminal).toBe('end');
+ expect(buffered.terminal).toBe('send');
+
+ // ⭐ Byte identity with the streamed leg (its leading res.write('')
+ // adapter nudge contributes zero bytes).
+ expect(buffered.sent).toBe(streamed.writes.join(''));
+ expect(decodeSseFrames(buffered.sent!)).toEqual(EXPECTED_FRAMES);
+
+ // The #9961 defect shape must not come back: the descriptor was
+ // json-serialized with `events` collapsed to {} and never drained.
+ expect(buffered.jsonBody).toBeUndefined();
+ expect(buffered.sent).not.toContain('"events"');
+ expect(a.drained.value).toBe(true);
+ expect(b.drained.value).toBe(true);
+
+ // Streaming headers relayed from the descriptor on both legs.
+ expect(streamed.headers['Content-Type']).toBe('text/event-stream');
+ expect(buffered.headers['Content-Type']).toBe('text/event-stream');
+ expect(buffered.status).toBe(200);
+ });
+
+ it('a mid-stream throw lands as the same event:error frame the streamed branch writes', async () => {
+ const boom = () => (async function* () {
+ yield STRING_FRAME;
+ throw new Error('adapter unplugged');
+ })();
+
+ const streamed = await drive(true, boom);
+ const buffered = await drive(false, boom);
+
+ // Streamed reference: frames then the error frame, then end().
+ const streamedBytes = streamed.writes.join('');
+ expect(streamed.terminal).toBe('end');
+ expect(streamedBytes).toContain('event: error');
+
+ // Buffered leg: byte-identical, delivered through send().
+ expect(buffered.terminal).toBe('send');
+ expect(buffered.sent).toBe(streamedBytes);
+ expect(buffered.sent).toContain('adapter unplugged');
+ expect(buffered.jsonBody).toBeUndefined();
+ });
+});

packages/runtime/src/dispatcher-plugin.ts

@@ -261,12 +261,30 @@ function mountRouteOnServer(
  }
  res.end();
  } else {
- // Fallback: collect events into array
- const events = [];
+ // [#9936] Buffered fallback — the IHttpResponse contract's
+ // own prescription (#3607, ADR-0076 OQ#10; the JSDoc on
+ // `write` in packages/spec/src/contracts/http-server.ts):
+ // a transport that omits the OPTIONAL `write`/`end`
+ // streaming surface receives the SAME SSE bytes, buffered
+ // and delivered through `send()` under the streaming
+ // headers already set above. A caller that asked for a
+ // stream parses this body with the same `data:`-line
+ // reader, frame for frame — the encoding ternary below is
+ // deliberately identical to the streamed branch's.
+ //
+ // This used to answer a bare `res.json({ events })`: a
+ // JSON dialect of the same frames that no SSE reader could
+ // decode (both shipped readers split raw bytes on
+ // newlines, and a JSON body contains none), off
+ // BaseResponseSchema besides. No shipped transport lacks
+ // `write`/`end`, so this branch is live only for an
+ // external `Runtime({ server })` transport — exactly the
+ // composition the contract note anticipates.
+ let buffered = '';
  for await (const event of result.events) {
-events.push(event);
+buffered += typeof event === 'string' ? event : `data: ${JSON.stringify(event)}\n\n`;
  }
- res.json({ events });
+ res.send(buffered);
  }
  } else {
  res.status(result.status);
@@ -385,6 +403,51 @@ function sendResultBase(
  })();
  return;
  }
+ if (isStream) {
+ // [#9961] Buffered fallback for a transport whose `res` cannot
+ // stream — the same #3607 / ADR-0076 OQ#10 contract
+ // prescription the route-wrapper branch applies (see
+ // `mountRouteOnServer`): drain the descriptor's AsyncIterable
+ // and deliver the identical SSE bytes through `send()` under
+ // the streaming headers. Frame encoding, the `null` skip and
+ // the trailing `event: error` frame all mirror the streamed
+ // branch above so the buffered body is byte-identical to what
+ // a streaming transport would have received.
+ //
+ // This used to fall through to `res.json(result.result)`,
+ // which serialized the descriptor itself: `JSON.stringify`
+ // collapses the `events` AsyncIterable to `{}`, so the caller
+ // got HTTP 200 with the payload gone and the iterable was
+ // never drained — silent total event loss.
+ res.status(typeof r.status === 'number' ? r.status : 200);
+ applySecurityHeaders();
+ if (r.headers && typeof r.headers === 'object') {
+ for (const [k, v] of Object.entries(r.headers)) {
+ res.header(k, String(v));
+ }
+ } else {
+ res.header('Content-Type', r.contentType || 'text/event-stream');
+ res.header('Cache-Control', 'no-cache');
+ res.header('Connection', 'keep-alive');
+ }
+ // Drained in the same detached shape as the streaming path —
+ // this function is synchronous by signature, and the response
+ // is delivered when the iterable settles.
+ (async () => {
+ let buffered = '';
+ try {
+ for await (const event of r.events as AsyncIterable<unknown>) {
+ if (event == null) continue;
+ buffered += typeof event === 'string' ? event : `data: ${JSON.stringify(event)}\n\n`;
+ }
+ } catch (streamErr) {
+ buffered += `event: error\ndata: ${JSON.stringify({ message: streamErr instanceof Error ? streamErr.message : String(streamErr) })}\n\n`;
+ } finally {
+ try { res.send(buffered); } catch { /* connection already gone */ }
+ }
+ })();
+ return;
+ }
  res.status(200);
  applySecurityHeaders();
  res.json(result.result);