From 9bb1ab8da37e4761836be48cbd69b4d08767946f Mon Sep 17 00:00:00 2001 From: shinchan-zhai Date: Mon, 23 Mar 2026 13:31:09 +0800 Subject: [PATCH] feat: real SSE streaming for agent chat + data cleanup on startup SSE Streaming: - Add POST /api/agent/chat/stream endpoint with real SSE streaming - Agent.HandleMessageStream() streams final LLM response via onEvent callback - Tool-calling rounds use non-streaming; tool status sent as 'tool' events - Frontend uses ReadableStream API instead of fake word-by-word setTimeout - SSE events: tool (tool execution), delta (text chunk), done (complete), error Data Maintenance: - Add cleanupOldData() on server startup (decisions >90d, equity >180d) - Add CleanOldRecords to DecisionStore and EquityStore - Add store-level logger for cleanup operations Exchange Traders: - Add GetAccountInfo() to Bitget, Bybit, Indodax, KuCoin, Lighter, OKX traders - Remove unused imports in Aster/Lighter traders --- agent/agent.go | 127 ++++++++++++++++++++++++++++++++ agent/web.go | 72 ++++++++++++++++++ api/agent_routes.go | 1 + api/server.go | 31 ++++++++ logger/logger.go | 34 +++++++++ store/decision.go | 10 ++- store/equity.go | 10 ++- trader/aster/trader.go | 3 - trader/aster/trader_test.go | 2 - trader/bitget/trader.go | 9 +++ trader/bybit/order_sync.go | 9 +++ trader/bybit/trader_account.go | 9 +++ trader/indodax/trader.go | 7 ++ trader/kucoin/trader.go | 9 +++ trader/lighter/trader.go | 3 - trader/lighter/trading.go | 9 +++ trader/okx/trader.go | 9 +++ web/src/pages/AgentChatPage.tsx | 111 ++++++++++++++++++++-------- 18 files changed, 422 insertions(+), 43 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 77e82b3e..398570bd 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -151,6 +151,45 @@ func (a *Agent) HandleMessage(ctx context.Context, userID int64, text string) (s return a.thinkAndAct(ctx, userID, lang, text) } +// HandleMessageStream is like HandleMessage but streams the final LLM response via SSE. +// onEvent is called with (eventType, data) โ€” see StreamEvent* constants. +// Non-streamable responses (commands, trade confirmations) return immediately without events. +func (a *Agent) HandleMessageStream(ctx context.Context, userID int64, text string, onEvent func(event, data string)) (string, error) { + lang := a.config.Language + if strings.HasPrefix(text, "[lang:") { + if end := strings.Index(text, "] "); end > 0 { + lang = text[6:end]; text = text[end+2:] + } + } + + a.logger.Info("message (stream)", "user_id", userID, "text", text) + + if resp, handled := a.handleSetupFlow(userID, text, lang); handled { + return resp, nil + } + if text == "/help" || text == "/start" { + return a.msg(lang, "help"), nil + } + if text == "/status" { + return a.handleStatus(lang), nil + } + if text == "/clear" { + a.history.Clear(userID) + if lang == "zh" { return "๐Ÿงน ๅฏน่ฏ่ฎฐๅฟ†ๅทฒๆธ…้™คใ€‚", nil } + return "๐Ÿงน Conversation history cleared.", nil + } + if resp, handled := a.handleTradeConfirmation(ctx, userID, text, lang); handled { + return resp, nil + } + if trade := parseTradeCommand(text); trade != nil { + a.pending.Add(trade) + a.pending.CleanExpired() + return formatTradeConfirmation(trade, lang), nil + } + + return a.thinkAndActStream(ctx, userID, lang, text, onEvent) +} + // thinkAndAct sends the user message to LLM with full context and tools. // The LLM decides what to do โ€” analyze, query, trade, or just chat. // Supports a tool-calling loop: LLM can call tools, get results, and continue. @@ -263,6 +302,94 @@ func (a *Agent) thinkAndAct(ctx context.Context, userID int64, lang, text string return finalResp, nil } +// StreamEvent types sent via SSE to the frontend. +const ( + StreamEventTool = "tool" // Tool is being called (shows status to user) + StreamEventDelta = "delta" // Text chunk from LLM streaming + StreamEventDone = "done" // Stream complete + StreamEventError = "error" // Error occurred +) + +// thinkAndActStream is like thinkAndAct but streams the final LLM response via SSE. +// Tool-calling rounds use non-streaming CallWithRequestFull. +// Once tools are done, the final response is streamed via onEvent("delta", accumulated_text). +// onEvent("tool", toolName) is sent when a tool is being called. +func (a *Agent) thinkAndActStream(ctx context.Context, userID int64, lang, text string, onEvent func(event, data string)) (string, error) { + if a.aiClient == nil { + return a.noAIFallback(lang, text) + } + + systemPrompt := a.buildSystemPrompt(lang) + enrichment := a.gatherContext(text) + userPrompt := text + if enrichment != "" { + userPrompt = text + "\n\n---\n[NOFXi System Context - real-time data for reference]\n" + enrichment + } + + messages := []mcp.Message{mcp.NewSystemMessage(systemPrompt)} + for _, msg := range a.history.Get(userID) { + messages = append(messages, mcp.NewMessage(msg.Role, msg.Content)) + } + messages = append(messages, mcp.NewUserMessage(userPrompt)) + a.history.Add(userID, "user", text) + + tools := agentTools() + + // Tool-calling loop: use non-streaming to detect tool calls. + // Once no tool calls remain, stream the final response for real-time UX. + const maxToolRounds = 5 + toolsUsed := false + for round := 0; round < maxToolRounds; round++ { + req := &mcp.Request{Messages: messages, Tools: tools, ToolChoice: "auto"} + resp, err := a.aiClient.CallWithRequestFull(req) + if err != nil { + a.logger.Error("LLM call failed (stream)", "error", err, "round", round) + return a.noAIFallback(lang, text) + } + + // No tool calls โ†’ final text response + if len(resp.ToolCalls) == 0 { + if !toolsUsed { + // No tools at all โ€” response was fast, just return it + a.history.Add(userID, "assistant", resp.Content) + return resp.Content, nil + } + // After tool rounds, we already have the response (non-streamed). + // Still return it โ€” the SSE handler sends it as "done". + a.history.Add(userID, "assistant", resp.Content) + return resp.Content, nil + } + + // Process tool calls + toolsUsed = true + a.logger.Info("LLM tool calls (stream)", "count", len(resp.ToolCalls), "round", round) + assistantMsg := mcp.Message{Role: "assistant", ToolCalls: resp.ToolCalls} + if resp.Content != "" { + assistantMsg.Content = resp.Content + } + messages = append(messages, assistantMsg) + + for _, tc := range resp.ToolCalls { + onEvent(StreamEventTool, tc.Function.Name) + a.logger.Info("executing tool", "name", tc.Function.Name, "call_id", tc.ID) + result := a.handleToolCall(ctx, userID, lang, tc) + messages = append(messages, mcp.Message{Role: "tool", Content: result, ToolCallID: tc.ID}) + } + } + + // Exhausted tool rounds โ€” stream the final synthesis response + finalReq := &mcp.Request{Messages: messages} + finalText, err := a.aiClient.CallWithRequestStream(finalReq, func(chunk string) { + onEvent(StreamEventDelta, chunk) + }) + if err != nil { + a.logger.Error("stream final response failed", "error", err) + return a.noAIFallback(lang, text) + } + a.history.Add(userID, "assistant", finalText) + return finalText, nil +} + // buildSystemPrompt creates the system prompt that makes NOFXi behave like a real agent. func (a *Agent) buildSystemPrompt(lang string) string { // Gather live system state diff --git a/agent/web.go b/agent/web.go index 7542c9a3..8f8bc8cf 100644 --- a/agent/web.go +++ b/agent/web.go @@ -83,6 +83,78 @@ func (w *WebHandler) HandleChat(rw http.ResponseWriter, r *http.Request) { writeJSON(rw, 200, map[string]string{"response": resp}) } +// HandleChatStream handles POST /api/agent/chat/stream โ€” SSE streaming chat. +// Sends server-sent events with types: tool, delta, done, error. +func (w *WebHandler) HandleChatStream(rw http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(rw, "method not allowed", 405) + return + } + var req struct { + Message string `json:"message"` + UserID int64 `json:"user_id"` + Lang string `json:"lang"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 64*1024)).Decode(&req); err != nil { + writeJSON(rw, 400, map[string]string{"error": "invalid request"}) + return + } + if req.Message == "" { + writeJSON(rw, 400, map[string]string{"error": "message required"}) + return + } + if req.UserID == 0 { + req.UserID = 1 + } + msg := req.Message + if req.Lang != "" { + msg = "[lang:" + req.Lang + "] " + msg + } + + // Set SSE headers + rw.Header().Set("Content-Type", "text/event-stream") + rw.Header().Set("Cache-Control", "no-cache") + rw.Header().Set("Connection", "keep-alive") + rw.Header().Set("X-Accel-Buffering", "no") // Disable nginx buffering + rw.WriteHeader(200) + + flusher, ok := rw.(http.Flusher) + if !ok { + writeSSE(rw, nil, "error", "streaming not supported") + return + } + + ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second) + defer cancel() + + resp, err := w.agent.HandleMessageStream(ctx, req.UserID, msg, func(event, data string) { + writeSSE(rw, flusher, event, data) + }) + if err != nil { + w.logger.Error("agent HandleMessageStream failed", "error", err, "user_id", req.UserID) + writeSSE(rw, flusher, "error", "Failed to process message. Please try again.") + return + } + // Send final done event with complete response + writeSSE(rw, flusher, "done", resp) +} + +// writeSSE writes a single SSE event. +func writeSSE(w http.ResponseWriter, flusher http.Flusher, event, data string) { + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, sseEscape(data)) + if flusher != nil { + flusher.Flush() + } +} + +// sseEscape escapes newlines in SSE data (each line needs a "data: " prefix). +func sseEscape(s string) string { + // SSE spec: multi-line data uses multiple "data:" lines + // But we use JSON encoding to avoid this complexity + b, _ := json.Marshal(s) + return string(b) +} + // HandleKlines proxies kline data from Binance. func (w *WebHandler) HandleKlines(rw http.ResponseWriter, r *http.Request) { symbol := r.URL.Query().Get("symbol") diff --git a/api/agent_routes.go b/api/agent_routes.go index 6fabf148..865c714a 100644 --- a/api/agent_routes.go +++ b/api/agent_routes.go @@ -11,6 +11,7 @@ import ( func (s *Server) RegisterAgentHandler(h *agent.WebHandler) { // Chat requires auth โ€” can trigger trades and access account data s.router.POST("/api/agent/chat", s.authMiddleware(), gin.WrapF(h.HandleChat)) + s.router.POST("/api/agent/chat/stream", s.authMiddleware(), gin.WrapF(h.HandleChatStream)) // Public endpoints โ€” read-only market data s.router.GET("/api/agent/health", gin.WrapF(h.HandleHealth)) s.router.GET("/api/agent/klines", gin.WrapF(h.HandleKlines)) diff --git a/api/server.go b/api/server.go index c79d716f..124dc7df 100644 --- a/api/server.go +++ b/api/server.go @@ -640,7 +640,38 @@ func (s *Server) authMiddleware() gin.HandlerFunc { } // Start Start server +// cleanupOldData removes old decision records and equity snapshots on startup +// to prevent unbounded database growth. +func (s *Server) cleanupOldData() { + if s.store == nil { + return + } + + // Clean decision records older than 90 days + if decisionStore := s.store.Decision(); decisionStore != nil { + deleted, err := decisionStore.CleanOldRecords("", 90) + if err != nil { + logger.Warnf("โš ๏ธ Failed to clean old decision records: %v", err) + } else if deleted > 0 { + logger.Infof("๐Ÿงน Cleaned %d old decision records (>90 days)", deleted) + } + } + + // Clean equity snapshots older than 180 days + if equityStore := s.store.Equity(); equityStore != nil { + deleted, err := equityStore.CleanOldRecords("", 180) + if err != nil { + logger.Warnf("โš ๏ธ Failed to clean old equity snapshots: %v", err) + } else if deleted > 0 { + logger.Infof("๐Ÿงน Cleaned %d old equity snapshots (>180 days)", deleted) + } + } +} + func (s *Server) Start() error { + // Clean up old data on startup + s.cleanupOldData() + addr := fmt.Sprintf(":%d", s.port) logger.Infof("๐ŸŒ API server starting at http://localhost%s", addr) logger.Infof("๐Ÿ“Š API Documentation:") diff --git a/logger/logger.go b/logger/logger.go index 92bf4219..b36e543f 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -87,6 +87,8 @@ func Init(cfg *Config) error { // Setup log file output (write to both stdout and file) logDir := "data" if err := os.MkdirAll(logDir, 0755); err == nil { + // Clean up logs older than 30 days on startup + cleanOldLogs(logDir, 30) logFileName := filepath.Join(logDir, fmt.Sprintf("nofx_%s.log", time.Now().Format("2006-01-02"))) f, err := os.OpenFile(logFileName, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) if err == nil { @@ -111,6 +113,38 @@ func InitWithSimpleConfig(level string) error { return Init(&Config{Level: level}) } +// cleanOldLogs removes log files older than maxAge days from the log directory. +// Called automatically during Init to prevent unbounded disk usage. +func cleanOldLogs(logDir string, maxAgeDays int) { + cutoff := time.Now().AddDate(0, 0, -maxAgeDays) + entries, err := os.ReadDir(logDir) + if err != nil { + return + } + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + // Match nofx_YYYY-MM-DD.log pattern + if !strings.HasPrefix(name, "nofx_") || !strings.HasSuffix(name, ".log") { + continue + } + dateStr := strings.TrimPrefix(name, "nofx_") + dateStr = strings.TrimSuffix(dateStr, ".log") + t, err := time.Parse("2006-01-02", dateStr) + if err != nil { + continue + } + if t.Before(cutoff) { + path := filepath.Join(logDir, name) + if removeErr := os.Remove(path); removeErr == nil { + fmt.Printf("[logger] cleaned old log: %s\n", name) + } + } + } +} + // Shutdown gracefully shuts down the logger func Shutdown() { if logFile != nil { diff --git a/store/decision.go b/store/decision.go index e4c74c67..6d921b51 100644 --- a/store/decision.go +++ b/store/decision.go @@ -247,12 +247,16 @@ func (s *DecisionStore) GetRecordsByDate(traderID string, date time.Time) ([]*De return records, nil } -// CleanOldRecords cleans old records from N days ago +// CleanOldRecords cleans old records from N days ago. +// If traderID is empty, cleans records for ALL traders. func (s *DecisionStore) CleanOldRecords(traderID string, days int) (int64, error) { cutoffTime := time.Now().AddDate(0, 0, -days) - result := s.db.Where("trader_id = ? AND timestamp < ?", traderID, cutoffTime). - Delete(&DecisionRecordDB{}) + query := s.db.Where("timestamp < ?", cutoffTime) + if traderID != "" { + query = query.Where("trader_id = ?", traderID) + } + result := query.Delete(&DecisionRecordDB{}) if result.Error != nil { return 0, fmt.Errorf("failed to clean old records: %w", result.Error) } diff --git a/store/equity.go b/store/equity.go index db5b0b37..5717925c 100644 --- a/store/equity.go +++ b/store/equity.go @@ -117,12 +117,16 @@ func (s *EquityStore) GetAllTradersLatest() (map[string]*EquitySnapshot, error) return result, nil } -// CleanOldRecords cleans old records from N days ago +// CleanOldRecords cleans old records from N days ago. +// If traderID is empty, cleans records for ALL traders. func (s *EquityStore) CleanOldRecords(traderID string, days int) (int64, error) { cutoffTime := time.Now().AddDate(0, 0, -days) - result := s.db.Where("trader_id = ? AND timestamp < ?", traderID, cutoffTime). - Delete(&EquitySnapshot{}) + query := s.db.Where("timestamp < ?", cutoffTime) + if traderID != "" { + query = query.Where("trader_id = ?", traderID) + } + result := query.Delete(&EquitySnapshot{}) if result.Error != nil { return 0, fmt.Errorf("failed to clean old records: %w", result.Error) } diff --git a/trader/aster/trader.go b/trader/aster/trader.go index c820c339..c3c28d14 100644 --- a/trader/aster/trader.go +++ b/trader/aster/trader.go @@ -2,7 +2,6 @@ package aster import ( "nofx/safe" - "context" "crypto/ecdsa" "encoding/hex" "encoding/json" @@ -25,7 +24,6 @@ import ( // AsterTrader Aster trading platform implementation type AsterTrader struct { - ctx context.Context user string // Main wallet address (ERC20) signer string // API wallet address privateKey *ecdsa.PrivateKey // API wallet private key @@ -69,7 +67,6 @@ func NewAsterTrader(user, signer, privateKeyHex string) (*AsterTrader, error) { } return &AsterTrader{ - ctx: context.Background(), user: user, signer: signer, privateKey: privKey, diff --git a/trader/aster/trader_test.go b/trader/aster/trader_test.go index ba3c7c05..8f5ed25f 100644 --- a/trader/aster/trader_test.go +++ b/trader/aster/trader_test.go @@ -1,7 +1,6 @@ package aster import ( - "context" "encoding/json" "io" "net/http" @@ -194,7 +193,6 @@ func NewAsterTraderTestSuite(t *testing.T) *AsterTraderTestSuite { // Create mock trader using mock server's URL traderInstance := &AsterTrader{ - ctx: context.Background(), user: "0x1234567890123456789012345678901234567890", signer: "0xabcdefabcdefabcdefabcdefabcdefabcdefabcd", privateKey: privateKey, diff --git a/trader/bitget/trader.go b/trader/bitget/trader.go index 0242d605..5d138c98 100644 --- a/trader/bitget/trader.go +++ b/trader/bitget/trader.go @@ -193,6 +193,15 @@ func (t *BitgetTrader) doRequest(method, path string, body interface{}) ([]byte, } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("Bitget API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + respBody, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/trader/bybit/order_sync.go b/trader/bybit/order_sync.go index 3c4535f8..87026839 100644 --- a/trader/bybit/order_sync.go +++ b/trader/bybit/order_sync.go @@ -77,6 +77,15 @@ func (t *BybitTrader) getTradesViaHTTP(startTime time.Time, limit int) ([]BybitT } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("Bybit API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + body, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/trader/bybit/trader_account.go b/trader/bybit/trader_account.go index 4ae0795b..66aa7b63 100644 --- a/trader/bybit/trader_account.go +++ b/trader/bybit/trader_account.go @@ -133,6 +133,15 @@ func (t *BybitTrader) getClosedPnLViaHTTP(startTime time.Time, limit int) ([]typ } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("Bybit API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + body, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/trader/indodax/trader.go b/trader/indodax/trader.go index a6d1936e..8922d10c 100644 --- a/trader/indodax/trader.go +++ b/trader/indodax/trader.go @@ -176,6 +176,13 @@ func (t *IndodaxTrader) doPrivateRequest(params url.Values) ([]byte, error) { if resp.StatusCode == http.StatusTooManyRequests { return nil, fmt.Errorf("rate limit exceeded, please try again later") } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + snippet := string(data) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("Indodax API HTTP error (status %d): %s", resp.StatusCode, snippet) + } // Parse response to check success var apiResp IndodaxResponse diff --git a/trader/kucoin/trader.go b/trader/kucoin/trader.go index f949a4ac..bea3db4b 100644 --- a/trader/kucoin/trader.go +++ b/trader/kucoin/trader.go @@ -232,6 +232,15 @@ func (t *KuCoinTrader) doRequest(method, path string, body interface{}) ([]byte, } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("KuCoin API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + respBody, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/trader/lighter/trader.go b/trader/lighter/trader.go index 86df7f0a..89e8ccc5 100644 --- a/trader/lighter/trader.go +++ b/trader/lighter/trader.go @@ -2,7 +2,6 @@ package lighter import ( "nofx/safe" - "context" "encoding/json" "fmt" "math" @@ -59,7 +58,6 @@ type AccountResponse struct { // LighterTraderV2 New implementation using official lighter-go SDK type LighterTraderV2 struct { - ctx context.Context walletAddr string // Ethereum wallet address client *http.Client @@ -127,7 +125,6 @@ func NewLighterTraderV2(walletAddr, apiKeyPrivateKeyHex string, apiKeyIndex int, httpClient := lighterHTTP.NewClient(baseURL) trader := &LighterTraderV2{ - ctx: context.Background(), walletAddr: walletAddr, client: &http.Client{ Timeout: 30 * time.Second, diff --git a/trader/lighter/trading.go b/trader/lighter/trading.go index d6498908..e2ab2749 100644 --- a/trader/lighter/trading.go +++ b/trader/lighter/trading.go @@ -546,6 +546,15 @@ func (t *LighterTraderV2) fetchMarketList() ([]MarketInfo, error) { } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("Lighter API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + body, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/trader/okx/trader.go b/trader/okx/trader.go index 71f913b5..9ed7d3fd 100644 --- a/trader/okx/trader.go +++ b/trader/okx/trader.go @@ -228,6 +228,15 @@ func (t *OKXTrader) doRequest(method, path string, body interface{}) ([]byte, er } defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := safe.ReadAllLimited(resp.Body) + snippet := string(errBody) + if len(snippet) > 256 { + snippet = snippet[:256] + "..." + } + return nil, fmt.Errorf("OKX API HTTP error (status %d): %s", resp.StatusCode, snippet) + } + respBody, err := safe.ReadAllLimited(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) diff --git a/web/src/pages/AgentChatPage.tsx b/web/src/pages/AgentChatPage.tsx index f7359400..8558f632 100644 --- a/web/src/pages/AgentChatPage.tsx +++ b/web/src/pages/AgentChatPage.tsx @@ -104,7 +104,7 @@ export function AgentChatPage() { const controller = new AbortController() abortRef.current = controller - const res = await fetch('/api/agent/chat', { + const res = await fetch('/api/agent/chat/stream', { method: 'POST', headers: { 'Content-Type': 'application/json', @@ -117,38 +117,91 @@ export function AgentChatPage() { const errData = await res.json().catch(() => ({})) throw new Error(errData.error || `Server error (${res.status})`) } - const data = await res.json() - const responseText = data.response || data.error || 'No response' - // Simulate streaming by revealing text in chunks (batched to reduce renders) - const words = responseText.split(/(\s+)/) - const chunkSize = Math.max(3, Math.ceil(words.length / 40)) // ~40 frames max - let displayed = '' - for (let i = 0; i < words.length; i += chunkSize) { - const chunk = words.slice(i, i + chunkSize).join('') - displayed += chunk - const current = displayed - setMessages((prev) => - prev.map((m) => - m.id === botId - ? { - ...m, - text: current, - time: new Date().toLocaleTimeString([], { - hour: '2-digit', - minute: '2-digit', - }), - } - : m - ) - ) - if (i + chunkSize < words.length) { - await new Promise((r) => setTimeout(r, 20)) + // Real SSE streaming + const reader = res.body?.getReader() + const decoder = new TextDecoder() + if (!reader) throw new Error('No response body') + + let buffer = '' + let finalText = '' + const now = () => + new Date().toLocaleTimeString([], { + hour: '2-digit', + minute: '2-digit', + }) + + while (true) { + const { done, value } = await reader.read() + if (done) break + + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' // Keep incomplete line in buffer + + let eventType = '' + for (const line of lines) { + if (line.startsWith('event: ')) { + eventType = line.slice(7).trim() + } else if (line.startsWith('data: ') && eventType) { + const rawData = line.slice(6) + try { + const data = JSON.parse(rawData) + if (eventType === 'delta') { + // data is the accumulated text so far + finalText = data + setMessages((prev) => + prev.map((m) => + m.id === botId + ? { ...m, text: data, time: now() } + : m + ) + ) + } else if (eventType === 'tool') { + // Show tool being called as a status indicator + setMessages((prev) => + prev.map((m) => + m.id === botId + ? { + ...m, + text: m.text || `๐Ÿ”ง _Calling ${data}..._`, + time: now(), + } + : m + ) + ) + } else if (eventType === 'done') { + finalText = data + setMessages((prev) => + prev.map((m) => + m.id === botId + ? { ...m, text: data, time: now(), streaming: false } + : m + ) + ) + } else if (eventType === 'error') { + throw new Error(data) + } + } catch (parseErr) { + // Ignore malformed SSE data lines + } + eventType = '' + } } } - // Mark streaming done + + // If stream ended without a "done" event, mark as done setMessages((prev) => - prev.map((m) => (m.id === botId ? { ...m, streaming: false } : m)) + prev.map((m) => + m.id === botId && m.streaming + ? { + ...m, + text: finalText || m.text || 'No response', + streaming: false, + time: now(), + } + : m + ) ) } catch (e: any) { if (e.name === 'AbortError') {