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..76b71bc7f9 --- /dev/null +++ b/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 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); + } + }); +}); + +// --------------------------------------------------------------------------- +// 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 52932bfd4a..1966f80c54 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); @@ -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) { + 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); 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': {}, }; /**