SSE, JSON frames, and bounded event streams

Reuse transport mechanics while preserving cancellation and backpressure; protocol parsing and business validation remain separate.

readSse parses standard SSE; encodeSseEvent/writeSseFrame only handle encoding and cancellable write backpressure, while HTTP responses, authentication, and lifecycle belong to the App framework. Browser consumption does not require a dedicated SDK backend.

streamEvents uses single-slot handoff, and producers must await emit rather than accumulating output without bound. When the consumer stops, the producer is notified, and unconsumed content is not treated as delivered.

scanJsonObjectFrames only extracts complete object frames and does not validate surrounding content; parseJsonFrames validates surrounding characters by maxFrames/maxExtraClosings and calls JSON.parse. They do not repair, truncate, or interpret business fields, and complete business structures are still validated by Schema.

Progressive examples for the same business case

Parse one SSE frame locally

Beginner · Node.js

TypeScript

import { Support } from '@jangts/agent-core';
const frame = Support.encodeSseEvent({ event: 'status', data: JSON.stringify({ phase: 'thinking' }) });
const body = new Response(frame).body!;
for await (const event of Support.readSse(body)) console.log(event.event, event.data);

JavaScript

import { Support } from '@jangts/agent-core';
const frame = Support.encodeSseEvent({ event: 'status', data: JSON.stringify({ phase: 'thinking' }) });
const body = new Response(frame).body;
for await (const event of Support.readSse(body))
    console.log(event.event, event.data);

Producer waits for each event delivery

Configuration · Node.js

TypeScript

import { Support } from '@jangts/agent-core';
const signal = new AbortController().signal;
const stream = Support.streamEvents<{ phase: string }>(async emit => {
  await emit({ phase: 'preparing' });
  await emit({ phase: 'complete' });
}, { signal });
for await (const event of stream) console.log(event.phase);

JavaScript

import { Support } from '@jangts/agent-core';
const signal = new AbortController().signal;
const stream = Support.streamEvents(async (emit) => {
    await emit({ phase: 'preparing' });
    await emit({ phase: 'complete' });
}, { signal });
for await (const event of stream)
    console.log(event.phase);

Explicitly limit JSON frame count and surrounding content

Extension · Node.js

TypeScript

import { Support } from '@jangts/agent-core';
const text = '{"kind":"reply","answer":"服务正常"}';
console.log(Support.scanJsonObjectFrames(text));
console.log(Support.parseJsonFrames(text, { maxFrames: 1, maxExtraClosings: 0 }));
const sent: string[] = [];
await Support.writeSseFrame({
  writer: { write(frame) { sent.push(frame); return true; }, async waitForDrain() {} },
  frame: Support.encodeSseEvent({ event: 'answer', data: text }),
}, { signal: new AbortController().signal });
console.log(sent);

JavaScript

import { Support } from '@jangts/agent-core';
const text = '{"kind":"reply","answer":"服务正常"}';
console.log(Support.scanJsonObjectFrames(text));
console.log(Support.parseJsonFrames(text, { maxFrames: 1, maxExtraClosings: 0 }));
const sent = [];
await Support.writeSseFrame({
    writer: { write(frame) { sent.push(frame); return true; }, async waitForDrain() { } },
    frame: Support.encodeSseEvent({ event: 'answer', data: text }),
}, { signal: new AbortController().signal });
console.log(sent);

Policies & implementations

• Single-slot handoff and backpressure in event streams