fix: improve log format and content for better debuggability

ysyneu · claude · ysyneu · commit c8022bd925e8 · 2026-03-24T20:35:06.000+08:00
Restore RFC3339 timestamps, use operation-specific log messages instead
of generic "executing task", standardize key names (server/tool), add
missing context to error logs (message type, raw_size, worknode_id),
and track MCP call goroutines for graceful shutdown.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/cmd/main.go b/cmd/main.go
@@ -244,11 +244,11 @@ func setupLogging(levelStr string) {
 	level := parseLogLevel(levelStr)
 	opts := &slog.HandlerOptions{
 		Level: level,
-		// Strip slog's time field — the process manager (systemd/journald)
-		// already prepends its own timestamp to every line.
 		ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr {
 			if a.Key == slog.TimeKey {
-				return slog.Attr{}
+				if t, ok := a.Value.Any().(time.Time); ok {
+					a.Value = slog.StringValue(t.Format(time.RFC3339))
+				}
 			}
 			return a
 		},
diff --git a/mcp/client.go b/mcp/client.go
@@ -44,24 +44,23 @@ func NewClientManager() *ClientManager {
 // GetSession returns or creates an MCP session for the given server.
 func (m *ClientManager) GetSession(ctx context.Context, server *protocol.MCPServerConfig, logger *slog.Logger) (*sdk_mcp.ClientSession, error) {
 	if logger == nil {
-		logger = slog.Default()
+		logger = slog.With("server", server.Name)
 	}
 
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
 	serverName := server.Name
 
-	logger.Debug("mcp get session",
-		"server_name", serverName,
+	logger.Debug("mcp resolving session",
 		"transport", server.Transport,
 		"url", server.URL,
 		"command", server.Command,
 	)
 
 	// Check if session exists and is still valid
 	if session, ok := m.sessions[serverName]; ok {
-		logger.Debug("mcp reusing existing session", "server_name", serverName)
+		logger.Debug("mcp reusing session")
 		return session, nil
 	}
 
@@ -77,51 +76,45 @@ func (m *ClientManager) GetSession(ctx context.Context, server *protocol.MCPServ
 
 	// Create transport
 	logger.Info("mcp creating transport",
-		"server_name", serverName,
 		"transport", server.Transport,
 		"url", server.URL,
 		"command", server.Command,
 	)
 
 	transport, err := createTransport(server)
 	if err != nil {
-		logger.Error("mcp create transport failed", "server_name", serverName, "error", err)
+		logger.Error("mcp transport creation failed", "error", err)
 		return nil, err
 	}
 
 	// Connect with timeout
-	logger.Info("mcp connecting to server", "server_name", serverName, "timeout", DefaultConnectTimeout)
+	logger.Info("mcp connecting", "timeout", DefaultConnectTimeout)
 
 	connectCtx, cancel := context.WithTimeout(ctx, DefaultConnectTimeout)
 	defer cancel()
 
 	session, err := client.Connect(connectCtx, transport, nil)
 	if err != nil {
-		logger.Error("mcp connect failed",
-			"server_name", serverName,
+		logger.Error("mcp connection failed",
 			"transport", server.Transport,
 			"url", server.URL,
 			"error", err,
 		)
 		return nil, fmt.Errorf("failed to connect to MCP server '%s': %w", serverName, err)
 	}
 
-	logger.Info("mcp connected successfully", "server_name", serverName)
+	logger.Info("mcp connected")
 	m.sessions[serverName] = session
 	return session, nil
 }
 
 // CallTool executes a single MCP tool call using the manager for session persistence.
 func (m *ClientManager) CallTool(ctx context.Context, server *protocol.MCPServerConfig, toolName string, arguments map[string]any, logger *slog.Logger) (*sdk_mcp.CallToolResult, error) {
 	if logger == nil {
-		logger = slog.Default()
+		logger = slog.With("server", server.Name, "tool", toolName)
 	}
 
-	logger.Debug("mcp call tool",
-		"server_name", server.Name,
-		"tool_name", toolName,
-		"arguments", arguments,
-	)
+	logger.Debug("mcp preparing tool call", "arguments", arguments)
 
 	session, err := m.GetSession(ctx, server, logger)
 	if err != nil {
@@ -131,34 +124,22 @@ func (m *ClientManager) CallTool(ctx context.Context, server *protocol.MCPServer
 	callCtx, callCancel := context.WithTimeout(ctx, DefaultCallTimeout)
 	defer callCancel()
 
-	logger.Debug("mcp calling tool", "server_name", server.Name, "tool_name", toolName, "timeout", DefaultCallTimeout)
+	logger.Debug("mcp invoking tool", "timeout", DefaultCallTimeout)
 
 	result, err := session.CallTool(callCtx, &sdk_mcp.CallToolParams{
 		Name:      toolName,
 		Arguments: arguments,
 	})
 	if err != nil {
-		logger.Error("mcp call tool failed",
-			"server_name", server.Name,
-			"tool_name", toolName,
-			"error", err,
-		)
+		logger.Error("mcp tool call failed", "error", err)
 		m.invalidateSession(server.Name)
 		return nil, fmt.Errorf("failed to call tool '%s': %w", toolName, err)
 	}
 
 	if result.IsError {
-		errorContent := ExtractContent(result)
-		logger.Warn("mcp tool returned error",
-			"server_name", server.Name,
-			"tool_name", toolName,
-			"error_content", errorContent,
-		)
+		logger.Warn("mcp tool returned error", "error_content", ExtractContent(result))
 	} else {
-		logger.Info("mcp call tool success",
-			"server_name", server.Name,
-			"tool_name", toolName,
-		)
+		logger.Info("mcp tool call succeeded")
 	}
 
 	return result, nil
diff --git a/mcp/transport.go b/mcp/transport.go
@@ -70,6 +70,23 @@ func (t *headerTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 		req.Header.Set(k, v)
 	}
 
+	// Reject standalone SSE GET immediately. Per MCP spec §2.2.3, servers
+	// that don't support standalone SSE streams MUST return 405. Some servers
+	// ignore the GET instead, causing the SDK to retry for 50+ seconds. We
+	// short-circuit with a synthetic 405 so Connect() completes quickly.
+	if req.Method == http.MethodGet && strings.Contains(req.Header.Get("Accept"), "text/event-stream") {
+		slog.Debug("mcp blocking standalone SSE GET",
+			"url", req.URL.String(),
+		)
+		return &http.Response{
+			StatusCode: http.StatusMethodNotAllowed,
+			Status:     "405 Method Not Allowed",
+			Header:     http.Header{},
+			Body:       http.NoBody,
+			Request:    req,
+		}, nil
+	}
+
 	slog.Debug("mcp http request",
 		"method", req.Method,
 		"url", req.URL.String(),
diff --git a/ws/client.go b/ws/client.go
@@ -210,6 +210,7 @@ func (c *Client) RunWithReconnect(ctx context.Context) error {
 
 			slog.Warn("connection failed, retrying",
 				"attempt", attempt,
+				"max_attempts", maxReconnectAttempts,
 				"delay", delay,
 				"error", err,
 			)
@@ -240,7 +241,8 @@ func (c *Client) RunWithReconnect(ctx context.Context) error {
 
 		// Run (blocking until disconnect)
 		if err := c.Run(ctx); err != nil {
-			slog.Warn("connection lost",
+			slog.Warn("connection lost, will reconnect",
+				"worknode_id", c.worknodeID,
 				"error", err,
 			)
 		}
@@ -356,14 +358,15 @@ func (c *Client) readLoop(ctx context.Context) error {
 		if err := json.Unmarshal(data, &msg); err != nil {
 			slog.Warn("failed to unmarshal message",
 				"error", err,
+				"raw_size", len(data),
 			)
 			continue
 		}
 
 		// Handle message in goroutine to not block read loop
 		go func(m protocol.Message) {
 			if err := c.handler(ctx, &m); err != nil {
-				slog.Error("failed to handle message",
+				slog.Error("message handling failed",
 					"type", m.Type,
 					"error", err,
 				)
@@ -387,20 +390,23 @@ func (c *Client) sendLoop(ctx context.Context) {
 			c.mu.Unlock()
 
 			if conn == nil {
+				slog.Warn("dropping message, not connected", "type", msg.Type)
 				continue
 			}
 
 			_ = conn.SetWriteDeadline(time.Now().Add(writeWait))
 			data, err := json.Marshal(msg)
 			if err != nil {
-				slog.Error("failed to marshal message",
+				slog.Error("failed to marshal outgoing message",
+					"type", msg.Type,
 					"error", err,
 				)
 				continue
 			}
 
 			if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
-				slog.Error("failed to send message",
+				slog.Error("failed to write message",
+					"type", msg.Type,
 					"error", err,
 				)
 			}
diff --git a/ws/handler.go b/ws/handler.go
@@ -150,15 +150,15 @@ func (h *Handler) executeTask(ctx context.Context, req *protocol.TaskRequestPayl
 		if err != nil {
 			return nil, fmt.Errorf("invalid read args: %w", err)
 		}
-		logger.Info("executing task", "path", args.Path)
+		logger.Info("reading file", "path", args.Path)
 		return h.ws.Read(ctx, args)
 
 	case protocol.TaskOpWrite:
 		args, err := parseArgs[protocol.WriteArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid write args: %w", err)
 		}
-		logger.Info("executing task", "path", args.Path)
+		logger.Info("writing file", "path", args.Path)
 		if err := h.ws.Write(ctx, args); err != nil {
 			return nil, err
 		}
@@ -169,50 +169,50 @@ func (h *Handler) executeTask(ctx context.Context, req *protocol.TaskRequestPayl
 		if err != nil {
 			return nil, fmt.Errorf("invalid list args: %w", err)
 		}
-		logger.Info("executing task", "path", args.Path, "recursive", args.Recursive)
+		logger.Info("listing directory", "path", args.Path, "recursive", args.Recursive)
 		return h.ws.List(ctx, args)
 
 	case protocol.TaskOpGlob:
 		args, err := parseArgs[protocol.GlobArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid glob args: %w", err)
 		}
-		logger.Info("executing task", "pattern", args.Pattern)
+		logger.Info("globbing files", "pattern", args.Pattern)
 		return h.ws.Glob(ctx, args)
 
 	case protocol.TaskOpGrep:
 		args, err := parseArgs[protocol.GrepArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid grep args: %w", err)
 		}
-		logger.Info("executing task", "pattern", args.Pattern, "include", args.Include)
+		logger.Info("grepping content", "pattern", args.Pattern, "include", args.Include)
 		return h.ws.Grep(ctx, args)
 
 	case protocol.TaskOpBash:
 		args, err := parseArgs[protocol.BashArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid bash args: %w", err)
 		}
-		logger.Info("executing task", "command", args.Command, "workdir", args.Workdir)
+		logger.Info("executing command", "command", args.Command, "workdir", args.Workdir)
 		return h.ws.Bash(ctx, args)
 
 	case protocol.TaskOpWebFetch:
 		args, err := parseArgs[protocol.WebFetchArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid webfetch args: %w", err)
 		}
-		logger.Info("executing task", "url", args.URL, "format", args.Format)
+		logger.Info("fetching url", "url", args.URL, "format", args.Format)
 		return h.ws.WebFetch(ctx, args)
 
 	case protocol.TaskOpMCPCall:
 		args, err := parseArgs[protocol.MCPCallArgs](req.Args)
 		if err != nil {
 			return nil, fmt.Errorf("invalid mcp_call args: %w", err)
 		}
-		logger.Info("executing task",
-			"server_name", args.Server.Name,
-			"server_transport", args.Server.Transport,
-			"tool_name", args.ToolName,
+		logger.Info("calling mcp tool",
+			"server", args.Server.Name,
+			"transport", args.Server.Transport,
+			"tool", args.ToolName,
 		)
 		return h.ws.MCPCall(ctx, args, logger)
 
@@ -221,9 +221,9 @@ func (h *Handler) executeTask(ctx context.Context, req *protocol.TaskRequestPayl
 		if err != nil {
 			return nil, fmt.Errorf("invalid mcp_list_tools args: %w", err)
 		}
-		logger.Info("executing task",
-			"server_name", args.Server.Name,
-			"server_transport", args.Server.Transport,
+		logger.Info("listing mcp tools",
+			"server", args.Server.Name,
+			"transport", args.Server.Transport,
 		)
 		return h.ws.MCPListTools(ctx, args, logger)
 
@@ -232,7 +232,7 @@ func (h *Handler) executeTask(ctx context.Context, req *protocol.TaskRequestPayl
 		if err != nil {
 			return nil, fmt.Errorf("invalid sync_skill args: %w", err)
 		}
-		logger.Info("executing task", "skill_name", args.SkillName, "skill_dir", args.SkillDir)
+		logger.Info("syncing skill", "skill", args.SkillName, "dir", args.SkillDir)
 		return h.ws.SyncSkill(ctx, args)
 
 	default:
@@ -281,8 +281,10 @@ func (h *Handler) handleTaskCancel(ctx context.Context, msg *protocol.Message) e
 	h.mu.RUnlock()
 
 	if ok {
-		slog.Info("cancelling task", "task_id", payload.TaskID)
+		slog.Info("cancelling task per request", "task_id", payload.TaskID)
 		cancel()
+	} else {
+		slog.Debug("cancel requested for completed or unknown task", "task_id", payload.TaskID)
 	}
 	return nil
 }
@@ -293,17 +295,27 @@ func (h *Handler) handleMCPCall(ctx context.Context, msg *protocol.Message) erro
 		return fmt.Errorf("failed to unmarshal mcp call: %w", err)
 	}
 
-	logger := slog.With("call_id", payload.CallID)
+	logger := slog.With(
+		"call_id", payload.CallID,
+		"server", payload.Server.Name,
+		"tool", payload.ToolName,
+	)
+
+	logger.Info("handling mcp call")
 
+	h.taskWg.Add(1)
 	go func() {
+		defer h.taskWg.Done()
 		result, err := h.ws.MCPCall(ctx, &protocol.MCPCallArgs{
 			Server:   payload.Server,
 			ToolName: payload.ToolName,
 			Args:     payload.Arguments,
 		}, logger)
 		if err != nil {
+			logger.Error("mcp call failed", "error", err)
 			h.sendMCPResult(payload.CallID, false, nil, err.Error())
 		} else {
+			logger.Info("mcp call completed")
 			h.sendMCPResult(payload.CallID, true, result, "")
 		}
 	}()

Original file line number	Diff line number	Diff line change
`@@ -210,6 +210,7 @@ func (c *Client) RunWithReconnect(ctx context.Context) error {`
`210`	`210`
`211`	`211`	`slog.Warn("connection failed, retrying",`
`212`	`212`	`"attempt", attempt,`
	`213`	`+ "max_attempts", maxReconnectAttempts,`
`213`	`214`	`"delay", delay,`
`214`	`215`	`"error", err,`
`215`	`216`	`)`
`@@ -240,7 +241,8 @@ func (c *Client) RunWithReconnect(ctx context.Context) error {`
`240`	`241`
`241`	`242`	`// Run (blocking until disconnect)`
`242`	`243`	`if err := c.Run(ctx); err != nil {`
`243`		`- slog.Warn("connection lost",`
	`244`	`+ slog.Warn("connection lost, will reconnect",`
	`245`	`+ "worknode_id", c.worknodeID,`
`244`	`246`	`"error", err,`
`245`	`247`	`)`
`246`	`248`	`}`
`@@ -356,14 +358,15 @@ func (c *Client) readLoop(ctx context.Context) error {`
`356`	`358`	`if err := json.Unmarshal(data, &msg); err != nil {`
`357`	`359`	`slog.Warn("failed to unmarshal message",`
`358`	`360`	`"error", err,`
	`361`	`+ "raw_size", len(data),`
`359`	`362`	`)`
`360`	`363`	`continue`
`361`	`364`	`}`
`362`	`365`
`363`	`366`	`// Handle message in goroutine to not block read loop`
`364`	`367`	`go func(m protocol.Message) {`
`365`	`368`	`if err := c.handler(ctx, &m); err != nil {`
`366`		`- slog.Error("failed to handle message",`
	`369`	`+ slog.Error("message handling failed",`
`367`	`370`	`"type", m.Type,`
`368`	`371`	`"error", err,`
`369`	`372`	`)`
`@@ -387,20 +390,23 @@ func (c *Client) sendLoop(ctx context.Context) {`
`387`	`390`	`c.mu.Unlock()`
`388`	`391`
`389`	`392`	`if conn == nil {`
	`393`	`+ slog.Warn("dropping message, not connected", "type", msg.Type)`
`390`	`394`	`continue`
`391`	`395`	`}`
`392`	`396`
`393`	`397`	`_ = conn.SetWriteDeadline(time.Now().Add(writeWait))`
`394`	`398`	`data, err := json.Marshal(msg)`
`395`	`399`	`if err != nil {`
`396`		`- slog.Error("failed to marshal message",`
	`400`	`+ slog.Error("failed to marshal outgoing message",`
	`401`	`+ "type", msg.Type,`
`397`	`402`	`"error", err,`
`398`	`403`	`)`
`399`	`404`	`continue`
`400`	`405`	`}`
`401`	`406`
`402`	`407`	`if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {`
`403`		`- slog.Error("failed to send message",`
	`408`	`+ slog.Error("failed to write message",`
	`409`	`+ "type", msg.Type,`
`404`	`410`	`"error", err,`
`405`	`411`	`)`
`406`	`412`	`}`