Function handleDirectTextStream

src/handler.ts:898–1139

(
    res: Response,
    cursorReq: CursorChatRequest,
    body: AnthropicRequest,
    log: RequestLogger,
    clientRequestedThinking: boolean,
    streamState: { blockIndex: number; textBlockStarted: boolean; thinkingEmitted: boolean },
)

}

async function handleDirectTextStream(
    res: Response,
    cursorReq: CursorChatRequest,
    body: AnthropicRequest,
    log: RequestLogger,
    clientRequestedThinking: boolean,
    streamState: { blockIndex: number; textBlockStarted: boolean; thinkingEmitted: boolean },
): Promise<void> {
    // ★ 流式保活：增量流式路径也需要 keepalive，防止 thinking 缓冲期间网关 504
    const keepaliveInterval = setInterval(() => {
        try {
            res.write(': keepalive\n\n');
            // @ts-expect-error flush exists on ServerResponse when compression is used
            if (typeof res.flush === 'function') res.flush();
        } catch { /* connection already closed, ignore */ }
    }, 15000);

    try {
    let activeCursorReq = cursorReq;
    let retryCount = 0;
    let finalRawResponse = '';
    let finalVisibleText = '';
    let finalThinkingContent = '';
    let cursorUsage: { inputTokens?: number; outputTokens?: number; totalTokens?: number } | undefined;
    let streamer = createIncrementalTextStreamer({
        warmupChars: 300,   // ★ 与工具模式对齐：前 300 chars 不释放，确保拒绝检测完成后再流
        transform: sanitizeResponse,
        isBlockedPrefix: (text) => isRefusal(text.substring(0, 300)),
    });

    const executeAttempt = async (): Promise<{
        rawResponse: string;
        visibleText: string;
        thinkingContent: string;
        streamer: ReturnType<typeof createIncrementalTextStreamer>;
    }> => {
        let rawResponse = '';
        let visibleText = '';
        let leadingBuffer = '';
        let leadingResolved = false;
        let thinkingContent = '';
        const attemptStreamer = createIncrementalTextStreamer({
            warmupChars: 300,   // ★ 与工具模式对齐
            transform: sanitizeResponse,
            isBlockedPrefix: (text) => isRefusal(text.substring(0, 300)),
        });

        const flushVisible = (chunk: string): void => {
            if (!chunk) return;
            visibleText += chunk;
            const delta = attemptStreamer.push(chunk);
            if (!delta) return;

            if (clientRequestedThinking && thinkingContent && !streamState.thinkingEmitted) {
                emitAnthropicThinkingBlock(res, streamState, thinkingContent);
            }
            writeAnthropicTextDelta(res, streamState, delta);
        };