feat: add GLMT streaming with real-time thinking blocks

- SSEParser/DeltaAccumulator for streaming state

- TTFB 5-20x faster (<500ms vs 2-10s)

- DoS protection: buffer limits, 120s timeout

- 51/51 tests passing (+25 streaming tests)
This commit is contained in:
kaitranntt committed 2025-11-11 03:47:30 -05:00
1 parent 657f99bfe8
commit 80f9cc644e
17 files changed
+1357 -71

No files matched your search

+178
View File
@@ -0,0 +1,178 @@
#!/usr/bin/env node
'use strict';
const DeltaAccumulator = require('../bin/delta-accumulator');
console.log('[TEST] DeltaAccumulator unit tests');
console.log('');
let passedTests = 0;
let failedTests = 0;
function test(name, fn) {
try {
fn();
console.log(`[PASS] ${name}`);
passedTests++;
} catch (error) {
console.log(`[FAIL] ${name}`);
console.log(` Error: ${error.message}`);
failedTests++;
}
}
function assert(condition, message) {
if (!condition) {
throw new Error(message || 'Assertion failed');
}
}
// Test: Initialization
test('Initial state', () => {
const acc = new DeltaAccumulator();
assert(acc.messageId.startsWith('msg_'), 'Message ID should start with msg_');
assert(acc.role === 'assistant', 'Default role should be assistant');
assert(acc.contentBlocks.length === 0, 'Should have no blocks initially');
assert(acc.currentBlockIndex === -1, 'Current index should be -1');
assert(!acc.messageStarted, 'Message should not be started');
assert(!acc.finalized, 'Should not be finalized');
});
// Test: Start thinking block
test('Start thinking block', () => {
const acc = new DeltaAccumulator();
const block = acc.startBlock('thinking');
assert(block.type === 'thinking', 'Block type should be thinking');
assert(block.index === 0, 'First block index should be 0');
assert(block.started === true, 'Block should be marked as started');
assert(block.stopped === false, 'Block should not be stopped');
assert(acc.contentBlocks.length === 1, 'Should have 1 block');
assert(acc.currentBlockIndex === 0, 'Current index should be 0');
});
// Test: Add delta to thinking block
test('Add delta to thinking block', () => {
const acc = new DeltaAccumulator();
acc.startBlock('thinking');
acc.addDelta('Hello ');
acc.addDelta('world');
const block = acc.getCurrentBlock();
assert(block.content === 'Hello world', `Expected 'Hello world', got '${block.content}'`);
assert(acc.thinkingBuffer === 'Hello world', 'Thinking buffer should match');
});
// Test: Start text block
test('Start text block', () => {
const acc = new DeltaAccumulator();
const block = acc.startBlock('text');
assert(block.type === 'text', 'Block type should be text');
assert(block.index === 0, 'First block index should be 0');
});
// Test: Add delta to text block
test('Add delta to text block', () => {
const acc = new DeltaAccumulator();
acc.startBlock('text');
acc.addDelta('Answer: ');
acc.addDelta('42');
const block = acc.getCurrentBlock();
assert(block.content === 'Answer: 42', `Expected 'Answer: 42', got '${block.content}'`);
assert(acc.textBuffer === 'Answer: 42', 'Text buffer should match');
});
// Test: Multiple blocks
test('Multiple blocks (thinking → text)', () => {
const acc = new DeltaAccumulator();
// Thinking block
acc.startBlock('thinking');
acc.addDelta('Analyzing...');
// Text block
acc.startBlock('text');
acc.addDelta('The answer is 42');
assert(acc.contentBlocks.length === 2, 'Should have 2 blocks');
assert(acc.currentBlockIndex === 1, 'Current index should be 1');
assert(acc.contentBlocks[0].type === 'thinking', 'First block should be thinking');
assert(acc.contentBlocks[1].type === 'text', 'Second block should be text');
assert(acc.contentBlocks[0].content === 'Analyzing...', 'Thinking content incorrect');
assert(acc.contentBlocks[1].content === 'The answer is 42', 'Text content incorrect');
});
// Test: Stop current block
test('Stop current block', () => {
const acc = new DeltaAccumulator();
acc.startBlock('thinking');
acc.addDelta('Done');
acc.stopCurrentBlock();
const block = acc.getCurrentBlock();
assert(block.stopped === true, 'Block should be marked as stopped');
});
// Test: Get current block (no blocks)
test('Get current block when none exist', () => {
const acc = new DeltaAccumulator();
const block = acc.getCurrentBlock();
assert(block === null, 'Should return null when no blocks');
});
// Test: Update usage statistics
test('Update usage statistics', () => {
const acc = new DeltaAccumulator();
acc.updateUsage({ prompt_tokens: 100, completion_tokens: 50 });
assert(acc.inputTokens === 100, `Expected 100, got ${acc.inputTokens}`);
assert(acc.outputTokens === 50, `Expected 50, got ${acc.outputTokens}`);
});
// Test: Get summary
test('Get summary', () => {
const acc = new DeltaAccumulator();
acc.model = 'glm-4.6';
acc.startBlock('thinking');
acc.addDelta('test');
acc.updateUsage({ prompt_tokens: 10, completion_tokens: 20 });
const summary = acc.getSummary();
assert(summary.messageId === acc.messageId, 'Message ID mismatch');
assert(summary.model === 'glm-4.6', 'Model mismatch');
assert(summary.role === 'assistant', 'Role mismatch');
assert(summary.blockCount === 1, `Expected 1 block, got ${summary.blockCount}`);
assert(summary.usage.input_tokens === 10, 'Input tokens mismatch');
assert(summary.usage.output_tokens === 20, 'Output tokens mismatch');
});
// Test: Thinking config preservation
test('Thinking config preservation', () => {
const config = { thinking: true, effort: 'high' };
const acc = new DeltaAccumulator(config);
assert(acc.thinkingConfig.thinking === true, 'Thinking config not preserved');
assert(acc.thinkingConfig.effort === 'high', 'Effort config not preserved');
});
// Test: Message lifecycle flags
test('Message lifecycle flags', () => {
const acc = new DeltaAccumulator();
assert(!acc.messageStarted, 'Should not be started initially');
acc.messageStarted = true;
assert(acc.messageStarted, 'Should be marked as started');
acc.finalized = true;
assert(acc.finalized, 'Should be marked as finalized');
});
// Test: Finish reason tracking
test('Finish reason tracking', () => {
const acc = new DeltaAccumulator();
assert(acc.finishReason === null, 'Finish reason should be null initially');
acc.finishReason = 'stop';
assert(acc.finishReason === 'stop', 'Finish reason should be updated');
});
console.log('');
console.log('═══════════════════════════════════════');
console.log(`TESTS: ${passedTests} passed, ${failedTests} failed`);
console.log('═══════════════════════════════════════');
if (failedTests > 0) {
process.exit(1);
}
+1 -1
View File
@@ -272,7 +272,7 @@ runner.test('disables streaming (not yet supported)', () => {
const { openaiRequest } = transformer.transformRequest(input);
assertEqual(openaiRequest.stream, false, 'stream should be disabled');
assertEqual(openaiRequest.stream, true, 'stream should be enabled when requested');
});
// Test 13: Debug mode disabled by default
+144
View File
@@ -0,0 +1,144 @@
#!/usr/bin/env node
'use strict';
const SSEParser = require('../bin/sse-parser');
console.log('[TEST] SSEParser unit tests');
console.log('');
let passedTests = 0;
let failedTests = 0;
function test(name, fn) {
try {
fn();
console.log(`[PASS] ${name}`);
passedTests++;
} catch (error) {
console.log(`[FAIL] ${name}`);
console.log(` Error: ${error.message}`);
failedTests++;
}
}
function assert(condition, message) {
if (!condition) {
throw new Error(message || 'Assertion failed');
}
}
// Test: Single event
test('Single event parsing', () => {
const parser = new SSEParser();
const events = parser.parse('data: {"test": "value"}\n\n');
assert(events.length === 1, `Expected 1 event, got ${events.length}`);
assert(events[0].data.test === 'value', `Expected test=value, got ${events[0].data.test}`);
});
// Test: Multiple events
test('Multiple events in one chunk', () => {
const parser = new SSEParser();
const events = parser.parse('data: {"a": 1}\n\ndata: {"b": 2}\n\n');
assert(events.length === 2, `Expected 2 events, got ${events.length}`);
assert(events[0].data.a === 1, 'First event data incorrect');
assert(events[1].data.b === 2, 'Second event data incorrect');
});
// Test: Split across chunks
test('Event split across chunks', () => {
const parser = new SSEParser();
const events1 = parser.parse('data: {"test":');
assert(events1.length === 0, 'Should not emit incomplete event');
const events2 = parser.parse('"value"}\n\n');
assert(events2.length === 1, `Expected 1 event after completion, got ${events2.length}`);
assert(events2[0].data.test === 'value', 'Split event data incorrect');
});
// Test: [DONE] marker
test('[DONE] marker detection', () => {
const parser = new SSEParser();
const events = parser.parse('data: [DONE]\n\n');
assert(events.length === 1, `Expected 1 event, got ${events.length}`);
assert(events[0].event === 'done', `Expected event=done, got ${events[0].event}`);
assert(events[0].data === null, 'Expected null data for [DONE]');
});
// Test: Mixed content and [DONE]
test('Mixed events with [DONE]', () => {
const parser = new SSEParser();
const events = parser.parse('data: {"msg": "hello"}\n\ndata: [DONE]\n\n');
assert(events.length === 2, `Expected 2 events, got ${events.length}`);
assert(events[0].data.msg === 'hello', 'First event data incorrect');
assert(events[1].event === 'done', 'Second event should be done');
});
// Test: Malformed JSON (should skip gracefully)
test('Malformed JSON handling', () => {
const parser = new SSEParser();
const events = parser.parse('data: {invalid json}\n\ndata: {"valid": true}\n\n');
// Should skip malformed, parse valid
assert(events.length === 1, `Expected 1 valid event, got ${events.length}`);
assert(events[0].data.valid === true, 'Valid event data incorrect');
});
// Test: Empty lines handling
test('Empty lines between events', () => {
const parser = new SSEParser();
const events = parser.parse('data: {"a": 1}\n\n\n\ndata: {"b": 2}\n\n');
assert(events.length === 2, `Expected 2 events, got ${events.length}`);
});
// Test: Event with ID field
test('Event with ID field', () => {
const parser = new SSEParser();
const events = parser.parse('id: 123\ndata: {"test": true}\n\n');
assert(events.length === 1, `Expected 1 event, got ${events.length}`);
assert(events[0].id === '123', `Expected id=123, got ${events[0].id}`);
assert(events[0].data.test === true, 'Event data incorrect');
});
// Test: Custom event type
test('Custom event type', () => {
const parser = new SSEParser();
const events = parser.parse('event: custom\ndata: {"test": true}\n\n');
assert(events.length === 1, `Expected 1 event, got ${events.length}`);
assert(events[0].event === 'custom', `Expected event=custom, got ${events[0].event}`);
});
// Test: Reset functionality
test('Parser reset', () => {
const parser = new SSEParser();
parser.parse('data: {"test": 1}\n\n');
assert(parser.eventCount === 1, 'Event count should be 1');
parser.reset();
assert(parser.eventCount === 0, 'Event count should be 0 after reset');
assert(parser.buffer === '', 'Buffer should be empty after reset');
});
// Test: Real Z.AI stream format
test('Real Z.AI stream format', () => {
const parser = new SSEParser();
const chunk = 'data: {"choices":[{"delta":{"role":"assistant","reasoning_content":"test"}}]}\n\n';
const events = parser.parse(chunk);
assert(events.length === 1, `Expected 1 event, got ${events.length}`);
assert(events[0].data.choices[0].delta.reasoning_content === 'test', 'Z.AI format parsing incorrect');
});
// Test: Multiple Z.AI deltas
test('Multiple Z.AI deltas', () => {
const parser = new SSEParser();
const chunk = 'data: {"choices":[{"delta":{"reasoning_content":"chunk1"}}]}\n\ndata: {"choices":[{"delta":{"reasoning_content":"chunk2"}}]}\n\n';
const events = parser.parse(chunk);
assert(events.length === 2, `Expected 2 events, got ${events.length}`);
assert(events[0].data.choices[0].delta.reasoning_content === 'chunk1', 'First delta incorrect');
assert(events[1].data.choices[0].delta.reasoning_content === 'chunk2', 'Second delta incorrect');
});
console.log('');
console.log('═══════════════════════════════════════');
console.log(`TESTS: ${passedTests} passed, ${failedTests} failed`);
console.log('═══════════════════════════════════════');
if (failedTests > 0) {
process.exit(1);
}
+230
View File
@@ -0,0 +1,230 @@
#!/usr/bin/env node
'use strict';
const https = require('https');
const API_KEY = process.env.Z_AI_API_KEY || process.env.ANTHROPIC_AUTH_TOKEN;
const MODEL = 'GLM-4.6';
if (!API_KEY || API_KEY === 'your-api-key-here') {
console.error('[ERROR] Z.AI API key not found');
console.error('[INFO] Set Z_AI_API_KEY or ANTHROPIC_AUTH_TOKEN environment variable');
console.error('[INFO] Example: export Z_AI_API_KEY=your-key-here');
process.exit(1);
}
// Test request with reasoning
const requestBody = JSON.stringify({
model: MODEL,
messages: [
{
role: 'user',
content: 'Solve this math problem step by step: What is 27 * 453? Show your reasoning process.'
}
],
stream: true,
reasoning: true,
reasoning_effort: 'medium',
max_tokens: 4096,
do_sample: true
});
const options = {
hostname: 'api.z.ai',
port: 443,
path: '/api/coding/paas/v4/chat/completions',
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${API_KEY}`,
'Content-Length': Buffer.byteLength(requestBody),
'User-Agent': 'CCS-GLMT-StreamingTest/1.0'
}
};
console.log('═══════════════════════════════════════════════════════════════');
console.log('Z.AI STREAMING BEHAVIOR TEST');
console.log('═══════════════════════════════════════════════════════════════');
console.log('[TEST] Initiating streaming request to Z.AI');
console.log('[TEST] Model:', MODEL);
console.log('[TEST] Stream: true');
console.log('[TEST] Reasoning: true');
console.log('[TEST] Reasoning Effort: medium');
console.log('[TEST] API Key:', API_KEY.substring(0, 10) + '...');
console.log('');
const startTime = Date.now();
let firstByteTime = null;
const req = https.request(options, (res) => {
console.log('[HTTP] Status:', res.statusCode);
console.log('[HTTP] Content-Type:', res.headers['content-type']);
console.log('');
if (res.statusCode !== 200) {
let errorBody = '';
res.on('data', chunk => errorBody += chunk);
res.on('end', () => {
console.error('[ERROR] Request failed:', res.statusCode, res.statusMessage);
console.error('[ERROR] Response:', errorBody);
process.exit(1);
});
return;
}
console.log('═══════════════════════════════════════════════════════════════');
console.log('SSE EVENTS (analyzing streaming behavior)');
console.log('═══════════════════════════════════════════════════════════════');
console.log('');
let buffer = '';
let eventCount = 0;
let reasoningEventCount = 0;
let reasoningInDelta = false;
let reasoningInMessage = false;
let totalReasoningChunks = [];
let textEventCount = 0;
res.on('data', (chunk) => {
if (!firstByteTime) {
firstByteTime = Date.now();
console.log('[TIMING] Time to first byte:', firstByteTime - startTime, 'ms');
console.log('');
}
buffer += chunk.toString();
const lines = buffer.split('\n');
// Keep incomplete line in buffer
buffer = lines.pop() || '';
lines.forEach(line => {
if (line.startsWith('data: ')) {
eventCount++;
const data = line.substring(6);
if (data === '[DONE]') {
console.log(`[EVENT ${eventCount}] [DONE]`);
console.log('');
return;
}
try {
const parsed = JSON.parse(data);
const choice = parsed.choices?.[0];
const delta = choice?.delta;
const message = choice?.message;
// Check for reasoning_content in delta
if (delta?.reasoning_content) {
reasoningInDelta = true;
reasoningEventCount++;
totalReasoningChunks.push(delta.reasoning_content);
console.log(`[EVENT ${eventCount}] *** REASONING IN DELTA ***`);
console.log(' Delta keys:', Object.keys(delta).join(', '));
console.log(' Reasoning chunk length:', delta.reasoning_content.length);
console.log(' Reasoning chunk preview:', JSON.stringify(delta.reasoning_content.substring(0, 80)));
console.log('');
}
// Check for reasoning_content in message (complete)
if (message?.reasoning_content) {
reasoningInMessage = true;
console.log(`[EVENT ${eventCount}] *** REASONING IN MESSAGE (COMPLETE) ***`);
console.log(' Message keys:', Object.keys(message).join(', '));
console.log(' Reasoning total length:', message.reasoning_content.length);
console.log(' Reasoning preview:', message.reasoning_content.substring(0, 100).replace(/\n/g, ' '));
console.log('');
}
// Check for text content in delta
if (delta?.content) {
textEventCount++;
if (textEventCount <= 3 || eventCount % 10 === 0) {
console.log(`[EVENT ${eventCount}] Text delta`);
console.log(' Content:', JSON.stringify(delta.content));
console.log('');
}
}
// Show finish_reason
if (choice?.finish_reason) {
console.log(`[EVENT ${eventCount}] Finish reason: ${choice.finish_reason}`);
console.log('');
}
// Show usage stats
if (parsed.usage) {
console.log(`[EVENT ${eventCount}] Usage:`, JSON.stringify(parsed.usage));
console.log('');
}
} catch (e) {
console.log(`[EVENT ${eventCount}] [PARSE ERROR]`, e.message);
console.log(' Raw:', data.substring(0, 100));
console.log('');
}
}
});
});
res.on('end', () => {
const totalTime = Date.now() - startTime;
console.log('═══════════════════════════════════════════════════════════════');
console.log('TEST RESULTS');
console.log('═══════════════════════════════════════════════════════════════');
console.log('');
console.log('[TIMING]');
console.log(' Time to first byte:', firstByteTime - startTime, 'ms');
console.log(' Total duration:', totalTime, 'ms');
console.log('');
console.log('[EVENTS]');
console.log(' Total events:', eventCount);
console.log(' Text content events:', textEventCount);
console.log(' Reasoning events:', reasoningEventCount);
console.log('');
console.log('[REASONING BEHAVIOR]');
console.log(' Reasoning in delta (incremental):', reasoningInDelta);
console.log(' Reasoning in message (complete):', reasoningInMessage);
if (totalReasoningChunks.length > 0) {
const totalReasoningLength = totalReasoningChunks.join('').length;
console.log(' Total reasoning chunks:', totalReasoningChunks.length);
console.log(' Total reasoning length:', totalReasoningLength);
}
console.log('');
console.log('═══════════════════════════════════════════════════════════════');
console.log('SCENARIO DETERMINATION');
console.log('═══════════════════════════════════════════════════════════════');
if (reasoningInDelta) {
console.log('');
console.log('[✓] SCENARIO A: Incremental Streaming');
console.log(' reasoning_content appears in delta objects');
console.log(' Multiple events contain reasoning chunks');
console.log(' RECOMMENDATION: Proceed with full streaming implementation');
} else if (reasoningInMessage) {
console.log('');
console.log('[!] SCENARIO B: Complete at End');
console.log(' reasoning_content only in final message object');
console.log(' No intermediate reasoning chunks');
console.log(' RECOMMENDATION: Hybrid streaming (text streams, reasoning buffered)');
} else {
console.log('');
console.log('[X] SCENARIO C: No Streaming Support');
console.log(' No reasoning_content in streaming mode');
console.log(' RECOMMENDATION: Keep buffered mode only');
}
console.log('');
console.log('═══════════════════════════════════════════════════════════════');
});
});
req.on('error', (error) => {
console.error('[ERROR]', error.message);
process.exit(1);
});
req.write(requestBody);
req.end();