From addfe554e294888bae74fdad093736e9a437c62c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 11:02:48 +0000 Subject: [PATCH 1/2] fix(runtime): buffer SSE frames through send() on write-less transports in the dispatcher route wrapper The mountRouteOnServer streaming branch's fallback answered a bare JSON { events } body no SSE reader could decode. It now implements the IHttpResponse streaming contract's prescription (#3607, ADR-0076 OQ#10): the same SSE bytes, buffered and delivered through send() under the streaming headers, pinned byte-identical to the streamed branch. The route-envelope declaration for dispatcher-plugin.ts goes 1 to 0 because the unenveloped JSON literal ceased to exist, not because it moved out of the scanner's sight. Family chain-head half; the sibling dispatch-result writer follows in the next commit. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WeN7F6jQFpcqW2BN56RdPa --- ...atcher-streaming-fallback-buffered-send.md | 5 + ...spatcher-plugin.streaming-fallback.test.ts | 221 ++++++++++++++++++ packages/runtime/src/dispatcher-plugin.ts | 26 ++- scripts/check-route-envelope.mjs | 20 +- 4 files changed, 262 insertions(+), 10 deletions(-) create mode 100644 .changeset/dispatcher-streaming-fallback-buffered-send.md create mode 100644 packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts diff --git a/.changeset/dispatcher-streaming-fallback-buffered-send.md b/.changeset/dispatcher-streaming-fallback-buffered-send.md new file mode 100644 index 0000000000..4f7a0c9d78 --- /dev/null +++ b/.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. diff --git a/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts b/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts new file mode 100644 index 0000000000..ceb5915738 --- /dev/null +++ b/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts @@ -0,0 +1,221 @@ +// 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 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>) { + // 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 = { 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; + writes: string[]; + sent?: string; + jsonBody?: unknown; + terminal?: 'json' | 'send' | 'end'; + } = { headers: {}, writes: [] }; + let resolveSettled!: () => void; + const settled = new Promise((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); + } + }); +}); + diff --git a/packages/runtime/src/dispatcher-plugin.ts b/packages/runtime/src/dispatcher-plugin.ts index 52932bfd4a..7edab52389 100644 --- a/packages/runtime/src/dispatcher-plugin.ts +++ b/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); diff --git a/scripts/check-route-envelope.mjs b/scripts/check-route-envelope.mjs index 6eeec6e736..4a4a8650af 100644 --- a/scripts/check-route-envelope.mjs +++ b/scripts/check-route-envelope.mjs @@ -704,7 +704,7 @@ const EXPRESS_RESPONSE_RECEIVERS = new Set(['res']); * receiver dialects. */ const IHTTP_ROUTE_MODULES = { - // Ten bodies. The two discovery bodies (`/.well-known/objectstack`, + // Nine bodies. The two discovery bodies (`/.well-known/objectstack`, // unconditional, and the REST-less `${prefix}/discovery` fallback) were // enveloped by #9813 under the #9436 maintainer ruling (2026-08-18, option // A, inherited with its reason intact: machine-read discovery bodies are @@ -713,11 +713,19 @@ const IHTTP_ROUTE_MODULES = { // these sites are not on). The two `{ success: false, error: buildApiError(…) }` // exits are conformant, and five relayed bodies (`result.body`, // `ANONYMOUS_DENY_BODY`, …) are deliberately invisible, as everywhere. - 'packages/runtime/src/dispatcher-plugin.ts': { - unenveloped: 1, - ratchet: '#9936 (envelope or rule on the SSE-fallback `{ events }` body)', - note: 'the streaming branch\'s JSON fallback — a transport whose `res` cannot stream gets the collected events as a bare `{ events }`, no `success` flag and the payload beside the envelope rather than under `data`. A different consumer population from the discovery bodies (callers that asked for an SSE stream), so #9813\'s inherited ruling does not reach it; #9936 carries the fork', - }, + // + // The tenth body — the SSE-fallback `res.json({ events })` this entry + // ratcheted at `unenveloped: 1` — was resolved by #9936 (2026-08-19, option + // B): the count goes to 0 because the JSON literal CEASED TO EXIST, not + // because it moved to a spelling this scanner cannot count. The fallback + // now implements the IHttpResponse contract's own prescription (#3607, + // ADR-0076 OQ#10): the same SSE frames, buffered and delivered through + // `send()` under the streaming headers — a body outside JSON entirely, + // pinned byte-identical to the streamed branch by + // `dispatcher-plugin.streaming-fallback.test.ts`. The sibling dispatch- + // result writer's write-less fallback (#9961) took the same shape; its old + // body was a RELAYED `res.json(result.result)` and thus never counted here. + 'packages/runtime/src/dispatcher-plugin.ts': {}, }; /** From aa0d23c8fb4c286da69a85a39814c7b301e121b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 11:02:59 +0000 Subject: [PATCH 2/2] fix(runtime): drain the SSE descriptor for write-less transports in the dispatch-result writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sendResultBase's write-less fallback fell through to res.json(result.result), serializing the stream descriptor with its events AsyncIterable collapsed to {} — HTTP 200, payload gone, iterable never drained. It now drains the frames and delivers the identical SSE bytes through send() under the streaming headers, mirroring the streamed branch's encoding (null skip and event: error frame included), pinned byte-identical to it. This body is a relayed res.json(...) invisible to check-route-envelope's counters by design, so no gate number moves for this half — the pin is the tripwire. Family member of the chain-head; same contract prescription (#3607, ADR-0076 OQ#10). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WeN7F6jQFpcqW2BN56RdPa --- ...spatcher-plugin.streaming-fallback.test.ts | 84 +++++++++++++++++++ packages/runtime/src/dispatcher-plugin.ts | 45 ++++++++++ 2 files changed, 129 insertions(+) diff --git a/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts b/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts index ceb5915738..76b71bc7f9 100644 --- a/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts +++ b/packages/runtime/src/dispatcher-plugin.streaming-fallback.test.ts @@ -219,3 +219,87 @@ describe('mountRouteOnServer write-less fallback (#9936)', () => { }); }); +// --------------------------------------------------------------------------- +// 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) { + 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(); + }); +}); diff --git a/packages/runtime/src/dispatcher-plugin.ts b/packages/runtime/src/dispatcher-plugin.ts index 7edab52389..1966f80c54 100644 --- a/packages/runtime/src/dispatcher-plugin.ts +++ b/packages/runtime/src/dispatcher-plugin.ts @@ -403,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) { + 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);