Server-Sent Events (SSE) দিয়ে আরও Reliable Streaming বানানো

id: 101
event: tool-start
data: {"type":"tool-start","toolName":"inspectLiveGitHubIssue","timestamp":1762014420000}

// src/lib/agent/sse-protocol.ts
export type AgentSSEPayload =
  | {
      type: 'phase';
      phase: 'routing' | 'synthesizing';
      detail: string;
      timestamp: number;
    }
  | {
      type: 'tool-start';
      toolCallId: string;
      toolName: string;
      args: unknown;
      timestamp: number;
    }
  | {
      type: 'tool-result';
      toolCallId: string;
      toolName: string;
      result: unknown;
      timestamp: number;
    }
  | {
      type: 'text-delta';
      delta: string;
    }
  | {
      type: 'heartbeat';
      timestamp: number;
    }
  | {
      type: 'error';
      message: string;
      retryable: boolean;
    }
  | {
      type: 'done';
      modelUsed: string;
      promptTokens: number;
      completionTokens: number;
      stepCount: number;
      latencyMs: number;
    };

export interface SSEStreamWriter {
  sendEvent: (payload: AgentSSEPayload) => void;
  close: () => void;
}

export function formatSSEFrame(id: number, payload: AgentSSEPayload): string {
  const serialized = JSON.stringify(payload);
  return `id: ${id}\nevent: ${payload.type}\ndata: ${serialized}\n\n`;
}

export function createReliableSSEStream(
  abortSignal: AbortSignal,
  executor: (writer: SSEStreamWriter) => Promise<void>,
  heartbeatIntervalMs = 3000
): ReadableStream<Uint8Array> {
  const encoder = new TextEncoder();
  let eventCounter = 0;
  let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
  let isClosed = false;

  function cleanupTimer() {
    if (heartbeatTimer) {
      clearInterval(heartbeatTimer);
      heartbeatTimer = null;
    }
  }

  return new ReadableStream<Uint8Array>({
    async start(controller) {
      const writer: SSEStreamWriter = {
        sendEvent(payload: AgentSSEPayload) {
          if (isClosed || abortSignal.aborted) return;
          eventCounter += 1;
          const frame = formatSSEFrame(eventCounter, payload);
          controller.enqueue(encoder.encode(frame));
        },
        close() {
          if (isClosed) return;
          isClosed = true;
          cleanupTimer();
          try {
            controller.close();
          } catch {
            // ক্লায়েন্ট কানেকশন আগেই বন্ধ হয়ে গেলে ইগনোর করবে
          }
        },
      };

      abortSignal.addEventListener('abort', () => {
        writer.close();
      });

      heartbeatTimer = setInterval(() => {
        writer.sendEvent({
          type: 'heartbeat',
          timestamp: Date.now(),
        });
      }, heartbeatIntervalMs);

      try {
        await executor(writer);
      } catch (err) {
        const message =
          err instanceof Error ? err.message : 'Unhandled SSE stream failure';
        writer.sendEvent({
          type: 'error',
          message,
          retryable: true,
        });
      } finally {
        writer.close();
      }
    },
    cancel() {
      isClosed = true;
      cleanupTimer();
    },
  });
}

// src/app/api/agent/sse/route.ts
import { NextRequest, NextResponse } from 'next/server';
import { z } from 'zod';
import { defaultDevPulseConfig } from '@/lib/agent/blueprint';
import { type ChatMessage } from '@/lib/agent/llm-client';
import { createAgentTextStream } from '@/lib/agent/stream-client';
import { createReliableSSEStream } from '@/lib/agent/sse-protocol';

export const maxDuration = 60;
export const dynamic = 'force-dynamic';

const sseRequestSchema = z.object({
  sessionId: z.string().min(8),
  modelId: z
    .enum(['gpt-4o', 'claude-3-5-sonnet-latest'])
    .default('claude-3-5-sonnet-latest'),
  messages: z
    .array(
      z.object({
        role: z.enum(['user', 'assistant']),
        content: z.string().trim().min(1).max(4000),
      })
    )
    .min(1),
});

export async function POST(req: NextRequest) {
  const body = await req.json().catch(() => null);
  const parsed = sseRequestSchema.safeParse(body);

  if (!parsed.success) {
    return NextResponse.json(
      { error: 'Invalid SSE request payload.' },
      { status: 400 }
    );
  }

  const { messages, modelId } = parsed.data;
  const startTime = Date.now();

  const stream = createReliableSSEStream(req.signal, async (writer) => {
    writer.sendEvent({
      type: 'phase',
      phase: 'routing',
      detail: `Connected to ${modelId} SSE stream...`,
      timestamp: Date.now(),
    });

    const result = createAgentTextStream(
      messages as ChatMessage[],
      {
        ...defaultDevPulseConfig,
        modelId,
      }
    );

    let stepCount = 1;

    for await (const part of result.fullStream) {
      if (req.signal.aborted) break;

      switch (part.type) {
        case 'tool-call': {
          writer.sendEvent({
            type: 'tool-start',
            toolCallId: part.toolCallId,
            toolName: part.toolName,
            args: part.args,
            timestamp: Date.now(),
          });
          break;
        }

        case 'tool-result': {
          writer.sendEvent({
            type: 'tool-result',
            toolCallId: part.toolCallId,
            toolName: part.toolName,
            result: part.result,
            timestamp: Date.now(),
          });
          writer.sendEvent({
            type: 'phase',
            phase: 'synthesizing',
            detail: `Completed ${part.toolName}. Synthesizing observation...`,
            timestamp: Date.now(),
          });
          stepCount += 1;
          break;
        }

        case 'text-delta': {
          writer.sendEvent({
            type: 'text-delta',
            delta: part.textDelta,
          });
          break;
        }

        case 'error': {
          const errMsg =
            part.error instanceof Error
              ? part.error.message
              : 'Model stream error occurred.';
          writer.sendEvent({
            type: 'error',
            message: errMsg,
            retryable: true,
          });
          break;
        }

        case 'finish': {
          writer.sendEvent({
            type: 'done',
            modelUsed: modelId,
            promptTokens: part.usage.promptTokens,
            completionTokens: part.usage.completionTokens,
            stepCount,
            latencyMs: Date.now() - startTime,
          });
          break;
        }
      }
    }
  });

  return new Response(stream, {
    status: 200,
    headers: {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      'X-Accel-Buffering': 'no',
    },
  });
}

// src/components/ReliableSSEChat.tsx
'use client';

import { useRef, useState } from 'react';
import { type AgentSSEPayload } from '@/lib/agent/sse-protocol';

export default function ReliableSSEChat() {
  const [prompt, setPrompt] = useState(
    'Inspect vercel/ai issue #402 and check our rate-limit docs for burst rules.'
  );
  const [answer, setAnswer] = useState('');
  const [statusText, setStatusText] = useState('Idle');
  const [toolLogs, setToolLogs] = useState<string[]>([]);
  const [lastHeartbeat, setLastHeartbeat] = useState<number | null>(null);
  const [isStreaming, setIsStreaming] = useState(false);
  const abortControllerRef = useRef<AbortController | null>(null);

  function handleIncomingSSEPayload(payload: AgentSSEPayload) {
    switch (payload.type) {
      case 'phase':
        setStatusText(payload.detail);
        break;
      case 'tool-start':
        setStatusText(`Running tool: ${payload.toolName}...`);
        setToolLogs((prev) => [
          ...prev,
          `⚡ Started ${payload.toolName} (${JSON.stringify(payload.args)})`,
        ]);
        break;
      case 'tool-result':
        setToolLogs((prev) => [
          ...prev,
          `✓ Finished ${payload.toolName}`,
        ]);
        break;
      case 'text-delta':
        setStatusText('Streaming answer...');
        setAnswer((prev) => prev + payload.delta);
        break;
      case 'heartbeat':
        setLastHeartbeat(payload.timestamp);
        break;
      case 'error':
        setStatusText(`Error: ${payload.message}`);
        break;
      case 'done':
        setStatusText(
          `Completed in ${payload.latencyMs}ms (${payload.promptTokens} in / ${payload.completionTokens} out tokens)`
        );
        break;
    }
  }

  async function startSSEStream(e: React.FormEvent) {
    e.preventDefault();
    if (!prompt.trim() || isStreaming) return;

    abortControllerRef.current?.abort();
    const controller = new AbortController();
    abortControllerRef.current = controller;

    setIsStreaming(true);
    setAnswer('');
    setToolLogs([]);
    setStatusText('Opening SSE connection...');

    try {
      const response = await fetch('/api/agent/sse', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          sessionId: 'sse-reliable-session-01',
          modelId: 'claude-3-5-sonnet-latest',
          messages: [{ role: 'user', content: prompt }],
        }),
        signal: controller.signal,
      });

      if (!response.ok || !response.body) {
        throw new Error(`HTTP ${response.status}: Failed to open SSE stream.`);
      }

      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let rawBuffer = '';

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        rawBuffer += decoder.decode(value, { stream: true });
        const frames = rawBuffer.split('\n\n');
        rawBuffer = frames.pop() ?? '';

        for (const frame of frames) {
          const dataLine = frame
            .split('\n')
            .find((line) => line.startsWith('data: '));

          if (!dataLine) continue;
          const jsonStr = dataLine.slice(6).trim();

          try {
            const payload = JSON.parse(jsonStr) as AgentSSEPayload;
            handleIncomingSSEPayload(payload);
          } catch {
            // অসম্পূর্ণ বা ভাঙা ফ্রেম ইগনোর করা হচ্ছে
          }
        }
      }
    } catch (err) {
      if (err instanceof Error && err.name === 'AbortError') {
        setStatusText('Stream cancelled by user.');
      } else {
        setStatusText(
          err instanceof Error ? err.message : 'Stream connection lost.'
        );
      }
    } finally {
      setIsStreaming(false);
    }
  }

  return (
    <div className='max-w-3xl mx-auto p-6 bg-slate-900 text-slate-100 rounded-xl border border-slate-800 space-y-4'>
      <div className='flex items-center justify-between border-b border-slate-800 pb-3'>
        <div>
          <h2 className='text-base font-semibold text-cyan-400'>
            DevPulse Resilient SSE Transport
          </h2>
          <p className='text-xs text-slate-400'>{statusText}</p>
        </div>
        {lastHeartbeat && (
          <span className='text-[11px] font-mono text-emerald-400 bg-emerald-950/50 px-2.5 py-1 rounded border border-emerald-800/60'>
            ♥ Heartbeat Active
          </span>
        )}
      </div>

      <form onSubmit={startSSEStream} className='flex gap-2'>
        <input
          value={prompt}
          onChange={(e) => setPrompt(e.target.value)}
          className='flex-1 px-3.5 py-2 rounded-lg bg-slate-950 border border-slate-700 text-sm'
        />
        {isStreaming ? (
          <button
            type='button'
            onClick={() => abortControllerRef.current?.abort()}
            className='px-4 py-2 rounded-lg bg-amber-600 hover:bg-amber-500 text-sm font-semibold'
          >
            Abort SSE
          </button>
        ) : (
          <button
            type='submit'
            className='px-4 py-2 rounded-lg bg-cyan-600 hover:bg-cyan-500 disabled:opacity-50 text-sm font-semibold'
          >
            Stream via SSE
          </button>
        )}
      </form>

      {toolLogs.length > 0 && (
        <div className='p-3 rounded-lg bg-slate-950 border border-purple-900/50 space-y-1 font-mono text-xs text-purple-300'>
          {toolLogs.map((log, i) => (
            <div key={i}>{log}</div>
          ))}
        </div>
      )}

      <div className='p-4 min-h-[140px] rounded-lg bg-slate-950 border border-slate-800 text-sm whitespace-pre-wrap leading-relaxed'>
        {answer || 'SSE frames will assemble here...'}
      </div>
    </div>
  );
}

rawBuffer += decoder.decode(value, { stream: true });
const frames = rawBuffer.split('\n\n');
rawBuffer = frames.pop() ?? '';