fix: SSE streaming for no-tool path, batch ticker API, security & cleanup

- 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
This commit is contained in:
shinchan-zhai
2026-03-23 14:01:11 +08:00
parent bd3d532239
commit 4c73a7960d
12 changed files with 135 additions and 34 deletions

View File

@@ -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 calledthe 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)

View File

@@ -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 }

View File

@@ -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 {

View File

@@ -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))
}

View File

@@ -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

View File

@@ -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
}

View File

@@ -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
}

View File

@@ -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
}

View File

@@ -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

View File

@@ -21,27 +21,23 @@ const SYMBOL_ICONS: Record<string, string> = {
export function MarketTicker() {
const [tickers, setTickers] = useState<Record<string, TickerData>>({})
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<string, TickerData> = {}
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)
}
}

View File

@@ -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)

View File

@@ -272,7 +272,7 @@ export function AgentChatPage() {
<div
style={{
display: 'flex',
height: 'calc(100vh - 64px)',
height: 'calc(100dvh - 64px)',
background: '#09090b',
overflow: 'hidden',
}}