From d6dafd830e4f75ce2580822fe1ee07a8a53be99e Mon Sep 17 00:00:00 2001 From: shinchan-zhai Date: Mon, 23 Mar 2026 14:01:11 +0800 Subject: [PATCH] fix: SSE streaming for no-tool path, batch ticker API, security & cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix thinkAndActStream: emit delta event for non-tool responses (was silent) - Add fallback to streaming on first LLM call failure - Add batch ticker endpoint /api/agent/tickers (3 API calls → 1) - Update MarketTicker to use batch endpoint, remove unused state - Warn loudly when JWT_SECRET uses default value - Fix 华为 mapping (was incorrectly mapped to Tencent) - Replace ioutil.ReadAll with safe.ReadAllLimited in nofxos client - Remove unused variable in hyperliquid trader_account - Wrap CoinAnk WS readers with safe.GoNamed for panic recovery - Use 100dvh for mobile viewport in AgentChatPage - console.log → console.warn for auth events --- agent/agent.go | 45 +++++++++++++++++--- agent/stock.go | 27 +++++++++--- agent/web.go | 52 +++++++++++++++++++++++ api/agent_routes.go | 1 + main.go | 7 ++- provider/coinank/coinank_api/depth_ws.go | 3 +- provider/coinank/coinank_api/kline_ws.go | 2 +- provider/nofxos/client.go | 4 +- trader/hyperliquid/trader_account.go | 2 - web/src/components/agent/MarketTicker.tsx | 22 ++++------ web/src/contexts/AuthContext.tsx | 2 +- web/src/pages/AgentChatPage.tsx | 2 +- 12 files changed, 135 insertions(+), 34 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 398570bd..11d0f349 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -335,27 +335,45 @@ func (a *Agent) thinkAndActStream(ctx context.Context, userID int64, lang, 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. + // Tool-calling loop with streaming: + // 1. Non-streaming call with tools to detect if LLM needs tools + // 2. If tools needed: execute them, loop back + // 3. When done (no more tools): stream the final response via SSE 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) + if round == 0 { + // First round failed — try streaming without tools as fallback + streamReq := &mcp.Request{Messages: messages} + streamText, streamErr := a.aiClient.CallWithRequestStream(streamReq, func(chunk string) { + onEvent(StreamEventDelta, chunk) + }) + if streamErr != nil { + return a.noAIFallback(lang, text) + } + a.history.Add(userID, "assistant", streamText) + return streamText, nil + } return a.noAIFallback(lang, text) } - // No tool calls → final text response + // No tool calls → done with tool loop if len(resp.ToolCalls) == 0 { if !toolsUsed { - // No tools at all — response was fast, just return it + // No tools were ever called — the non-streaming probe already has the answer. + // Emit as a single delta so frontend renders it immediately. + onEvent(StreamEventDelta, resp.Content) 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". + // Tools were used in previous rounds, LLM gave final answer without streaming. + // This shouldn't normally happen (we break and stream below), but handle it. + onEvent(StreamEventDelta, resp.Content) a.history.Add(userID, "assistant", resp.Content) return resp.Content, nil } @@ -375,9 +393,22 @@ func (a *Agent) thinkAndActStream(ctx context.Context, userID int64, lang, text result := a.handleToolCall(ctx, userID, lang, tc) messages = append(messages, mcp.Message{Role: "tool", Content: result, ToolCallID: tc.ID}) } + + // After tool execution, stream the next LLM response for real-time UX. + // Omit tools so LLM can't start another tool round — it must produce text. + streamReq := &mcp.Request{Messages: messages} + streamText, streamErr := a.aiClient.CallWithRequestStream(streamReq, func(chunk string) { + onEvent(StreamEventDelta, chunk) + }) + if streamErr != nil { + a.logger.Error("stream post-tool response failed", "error", streamErr, "round", round) + return a.noAIFallback(lang, text) + } + a.history.Add(userID, "assistant", streamText) + return streamText, nil } - // Exhausted tool rounds — stream the final synthesis response + // Exhausted all tool rounds — stream the final synthesis response finalReq := &mcp.Request{Messages: messages} finalText, err := a.aiClient.CallWithRequestStream(finalReq, func(chunk string) { onEvent(StreamEventDelta, chunk) diff --git a/agent/stock.go b/agent/stock.go index 8b2d2b04..250e7d8c 100644 --- a/agent/stock.go +++ b/agent/stock.go @@ -14,6 +14,17 @@ import ( "golang.org/x/text/transform" ) +// stockHTTPClient is a shared HTTP client for stock API requests. +// Reused across calls for connection pooling. +var stockHTTPClient = &http.Client{ + Timeout: 10 * time.Second, + Transport: &http.Transport{ + MaxIdleConns: 10, + MaxIdleConnsPerHost: 5, + IdleConnTimeout: 90 * time.Second, + }, +} + // StockQuote holds real-time stock data. type StockQuote struct { Name string @@ -54,7 +65,7 @@ var knownStocks = map[string]string{ "小米": "hk01810", "京东": "hk09618", "网易": "hk09999", "百度": "hk09888", "快手": "hk01024", "哔哩哔哩": "hk09626", "理想汽车": "hk02015", "蔚来": "hk09866", "小鹏汽车": "hk09868", - "华为": "hk00700", // fallback to tencent for now + // 华为 is not publicly listed — removed incorrect Tencent fallback // 美股 "苹果": "gb_aapl", "特斯拉": "gb_tsla", "英伟达": "gb_nvda", "微软": "gb_msft", "谷歌": "gb_googl", "亚马逊": "gb_amzn", @@ -127,13 +138,16 @@ func searchStock(keyword string) ([]SearchResult, error) { req, _ := http.NewRequest("GET", u, nil) req.Header.Set("Referer", "https://finance.sina.com.cn") - client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Do(req) + resp, err := stockHTTPClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("stock search API returned status %d", resp.StatusCode) + } + reader := transform.NewReader(io.LimitReader(resp.Body, 256*1024), simplifiedchinese.GBK.NewDecoder()) body, err := safe.ReadAllLimited(reader) if err != nil { @@ -283,11 +297,14 @@ func fetchStockQuote(code string) (*StockQuote, error) { req, _ := http.NewRequest("GET", url, nil) req.Header.Set("Referer", "https://finance.sina.com.cn") - client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Do(req) + resp, err := stockHTTPClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("stock quote API returned status %d", resp.StatusCode) + } + reader := transform.NewReader(io.LimitReader(resp.Body, 256*1024), simplifiedchinese.GBK.NewDecoder()) body, err := safe.ReadAllLimited(reader) if err != nil { return nil, err } diff --git a/agent/web.go b/agent/web.go index 8f8bc8cf..b4b42132 100644 --- a/agent/web.go +++ b/agent/web.go @@ -187,6 +187,58 @@ func (w *WebHandler) HandleTicker(rw http.ResponseWriter, r *http.Request) { proxyBinance(rw, fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbol=%s", symbol)) } +// HandleTickers handles GET /api/agent/tickers?symbols=BTCUSDT,ETHUSDT,SOLUSDT +// Batch endpoint: fetches multiple tickers in one API call to Binance. +func (w *WebHandler) HandleTickers(rw http.ResponseWriter, r *http.Request) { + symbolsParam := r.URL.Query().Get("symbols") + if symbolsParam == "" { + symbolsParam = "BTCUSDT,ETHUSDT,SOLUSDT" + } + + // Validate and build JSON array of symbols + symbols := []string{} + for _, s := range splitAndTrim(symbolsParam) { + if validSymbolRe.MatchString(s) { + symbols = append(symbols, `"`+s+`"`) + } + } + if len(symbols) == 0 { + writeJSON(rw, 400, map[string]string{"error": "no valid symbols"}) + return + } + if len(symbols) > 20 { + writeJSON(rw, 400, map[string]string{"error": "max 20 symbols"}) + return + } + + // Binance supports batch ticker: /fapi/v1/ticker/24hr with multiple symbols + url := fmt.Sprintf("https://fapi.binance.com/fapi/v1/ticker/24hr?symbols=[%s]", joinStrings(symbols, ",")) + proxyBinance(rw, url) +} + +// splitAndTrim splits a comma-separated string and trims whitespace. +func splitAndTrim(s string) []string { + parts := []string{} + for _, p := range regexp.MustCompile(`\s*,\s*`).Split(s, -1) { + if p != "" { + parts = append(parts, p) + } + } + return parts +} + +// joinStrings joins strings with a separator. +func joinStrings(ss []string, sep string) string { + result := "" + for i, s := range ss { + if i > 0 { + result += sep + } + result += s + } + return result +} + func proxyBinance(rw http.ResponseWriter, url string) { resp, err := binanceClient.Get(url) if err != nil { diff --git a/api/agent_routes.go b/api/agent_routes.go index 865c714a..ea9f277c 100644 --- a/api/agent_routes.go +++ b/api/agent_routes.go @@ -16,4 +16,5 @@ func (s *Server) RegisterAgentHandler(h *agent.WebHandler) { s.router.GET("/api/agent/health", gin.WrapF(h.HandleHealth)) s.router.GET("/api/agent/klines", gin.WrapF(h.HandleKlines)) s.router.GET("/api/agent/ticker", gin.WrapF(h.HandleTicker)) + s.router.GET("/api/agent/tickers", gin.WrapF(h.HandleTickers)) } diff --git a/main.go b/main.go index eddb7419..610e99ed 100644 --- a/main.go +++ b/main.go @@ -89,7 +89,12 @@ func main() { // Set JWT secret auth.SetJWTSecret(cfg.JWTSecret) - logger.Info("🔑 JWT secret configured") + if cfg.JWTSecret == "default-jwt-secret-change-in-production" { + logger.Warn("⚠️ JWT_SECRET is using the default value — this is INSECURE for production!") + logger.Warn("⚠️ Set JWT_SECRET environment variable to a strong random secret") + } else { + logger.Info("🔑 JWT secret configured") + } // WebSocket market monitor is NO LONGER USED // All K-line data now comes from CoinAnk API instead of Binance WebSocket cache diff --git a/provider/coinank/coinank_api/depth_ws.go b/provider/coinank/coinank_api/depth_ws.go index 679a38b6..208f66d0 100644 --- a/provider/coinank/coinank_api/depth_ws.go +++ b/provider/coinank/coinank_api/depth_ws.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "nofx/provider/coinank/coinank_enum" + "nofx/safe" "golang.org/x/net/websocket" ) @@ -79,7 +80,7 @@ func depth_ws(ctx context.Context) (*websocket.Conn, <-chan *WsResult[DepthV3], return nil, nil, err } ch := make(chan *WsResult[DepthV3], 1024) - go depth_read(conn, ch) + safe.GoNamed("coinank-depth-ws-reader", func() { depth_read(conn, ch) }) return conn, ch, nil } diff --git a/provider/coinank/coinank_api/kline_ws.go b/provider/coinank/coinank_api/kline_ws.go index d0a0db39..bd7be8df 100644 --- a/provider/coinank/coinank_api/kline_ws.go +++ b/provider/coinank/coinank_api/kline_ws.go @@ -86,7 +86,7 @@ func ws(ctx context.Context) (*websocket.Conn, <-chan string, error) { return nil, nil, err } ch := make(chan string, 1024) - go read(conn, ch) + safe.GoNamed("coinank-ws-reader", func() { read(conn, ch) }) return conn, ch, nil } diff --git a/provider/nofxos/client.go b/provider/nofxos/client.go index a3a99a00..9d8bc32a 100644 --- a/provider/nofxos/client.go +++ b/provider/nofxos/client.go @@ -4,8 +4,8 @@ package nofxos import ( - "io/ioutil" "net/http" + "nofx/safe" "nofx/security" "strings" "sync" @@ -108,7 +108,7 @@ func (c *Client) doRequest(endpoint string) ([]byte, error) { } defer resp.Body.Close() - body, err := ioutil.ReadAll(resp.Body) + body, err := safe.ReadAllLimited(resp.Body, 0) if err != nil { return nil, err } diff --git a/trader/hyperliquid/trader_account.go b/trader/hyperliquid/trader_account.go index 92eb1f88..8b63b60a 100644 --- a/trader/hyperliquid/trader_account.go +++ b/trader/hyperliquid/trader_account.go @@ -45,8 +45,6 @@ func (t *HyperliquidTrader) GetBalance() (map[string]interface{}, error) { // Step 3: Dynamically select correct summary based on margin mode (CrossMarginSummary or MarginSummary) var accountValue, totalMarginUsed float64 var summaryType string - var summary interface{} - _, _ = summaryType, summary if t.isCrossMargin { // Cross margin mode: use CrossMarginSummary diff --git a/web/src/components/agent/MarketTicker.tsx b/web/src/components/agent/MarketTicker.tsx index 72aa3853..5e6e7430 100644 --- a/web/src/components/agent/MarketTicker.tsx +++ b/web/src/components/agent/MarketTicker.tsx @@ -21,27 +21,23 @@ const SYMBOL_ICONS: Record = { export function MarketTicker() { const [tickers, setTickers] = useState>({}) const [loading, setLoading] = useState(true) - const [, setRefreshing] = useState(false) const fetchTickers = async () => { try { - const results = await Promise.all( - SYMBOLS.map(async (symbol) => { - const res = await fetch(`/api/agent/ticker?symbol=${symbol}`) - const data = await res.json() - return { symbol, ...data } - }) - ) + // Batch fetch: single API call for all symbols + const res = await fetch(`/api/agent/tickers?symbols=${SYMBOLS.join(',')}`) + const data = await res.json() const map: Record = {} - results.forEach((r) => { - if (r.lastPrice) map[r.symbol] = r - }) + if (Array.isArray(data)) { + data.forEach((r: TickerData) => { + if (r.lastPrice && r.symbol) map[r.symbol] = r + }) + } setTickers(map) } catch { - // ignore + // ignore — will retry on next interval } finally { setLoading(false) - setRefreshing(false) } } diff --git a/web/src/contexts/AuthContext.tsx b/web/src/contexts/AuthContext.tsx index 3132c68f..69b82f7d 100644 --- a/web/src/contexts/AuthContext.tsx +++ b/web/src/contexts/AuthContext.tsx @@ -75,7 +75,7 @@ export function AuthProvider({ children }: { children: React.ReactNode }) { // Listen for unauthorized events from httpClient (401 responses) useEffect(() => { const handleUnauthorized = () => { - console.log('Unauthorized event received - clearing auth state') + console.warn('Unauthorized event received - clearing auth state') // Clear auth state when 401 is detected setUser(null) setToken(null) diff --git a/web/src/pages/AgentChatPage.tsx b/web/src/pages/AgentChatPage.tsx index 8558f632..7fa32c5f 100644 --- a/web/src/pages/AgentChatPage.tsx +++ b/web/src/pages/AgentChatPage.tsx @@ -272,7 +272,7 @@ export function AgentChatPage() {