1package lsp
2
3import (
4 "bufio"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "time"
17)
18
19type rpcError struct {
20 Code int `json:"code"`
21 Message string `json:"message"`
22 Data any `json:"data,omitempty"`
23}
24
25type rpcResponse struct {
26 Result json.RawMessage
27 Error *rpcError
28}
29
30type rpcEnvelope struct {
31 JSONRPC string `json:"jsonrpc"`
32 ID json.RawMessage `json:"id"`
33 Method string `json:"method"`
34 Params json.RawMessage `json:"params"`
35 Result json.RawMessage `json:"result"`
36 Error *rpcError `json:"error"`
37}
38
39// ClientOptions configures notifications and the small set of client-side
40// handlers required by real language servers.
41type ClientOptions struct {
42 Settings map[string]any
43 InitializationOptions any
44 OnDiagnostics func([]Diagnostic)
45 OnLog func(string)
46 OnProgress func(string, json.RawMessage)
47 ApplyEdit func(WorkspaceEdit) error
48}
49
50// Client is one stdio JSON-RPC LSP process. A Client is safe for concurrent
51// requests; the process itself is shared by all open documents in its root.
52type Client struct {
53 Spec ServerConfig
54 Root string
55 Command string
56 cmd *exec.Cmd
57 stdin io.WriteCloser
58 stdout io.ReadCloser
59 stderr io.ReadCloser
60
61 writeMu sync.Mutex
62 applyEditMu sync.Mutex
63 mu sync.Mutex
64 pending map[int64]chan rpcResponse
65 nextID int64
66 closed bool
67 closeOnce sync.Once
68 done chan struct{}
69 doneOnce sync.Once
70 readErr atomic.Value // string; one concrete type keeps atomic.Value safe
71
72 options ClientOptions
73 capMu sync.RWMutex
74 caps json.RawMessage
75 docsMu sync.Mutex
76 docs map[string]int
77}
78
79// NewClient starts a stdio server but does not initialize it. This separation
80// is convenient for tests and lets callers choose an initialization timeout.
81func NewClient(spec ServerConfig, root string) (*Client, error) {
82 return NewClientWithOptions(spec, root, ClientOptions{})
83}
84
85// NewClientWithOptions starts a client with notification callbacks installed.
86func NewClientWithOptions(spec ServerConfig, root string, options ClientOptions) (*Client, error) {
87 if strings.EqualFold(spec.Kind, "cli") {
88 return nil, fmt.Errorf("%s is a CLI provider, not an LSP server", spec.ID)
89 }
90 root, err := filepath.Abs(root)
91 if err != nil {
92 return nil, err
93 }
94 command, err := ResolveCommand(root, spec.Command)
95 if err != nil {
96 return nil, fmt.Errorf("%s: %w", spec.ID, err)
97 }
98 cmd := exec.Command(command, spec.Args...)
99 cmd.WaitDelay = 2 * time.Second
100 cmd.Dir = root
101 cmd.Env = mergedEnvironment(spec.Env)
102 stdin, err := cmd.StdinPipe()
103 if err != nil {
104 return nil, fmt.Errorf("LSP stdin: %w", err)
105 }
106 stdout, err := cmd.StdoutPipe()
107 if err != nil {
108 _ = stdin.Close()
109 return nil, fmt.Errorf("LSP stdout: %w", err)
110 }
111 stderr, err := cmd.StderrPipe()
112 if err != nil {
113 _ = stdin.Close()
114 _ = stdout.Close()
115 return nil, fmt.Errorf("LSP stderr: %w", err)
116 }
117 if err := cmd.Start(); err != nil {
118 _ = stdin.Close()
119 _ = stdout.Close()
120 _ = stderr.Close()
121 return nil, fmt.Errorf("start %s: %w", spec.ID, err)
122 }
123 c := &Client{
124 Spec: spec, Root: root, Command: command, cmd: cmd,
125 stdin: stdin, stdout: stdout, stderr: stderr,
126 pending: make(map[int64]chan rpcResponse), done: make(chan struct{}),
127 options: options, docs: make(map[string]int),
128 }
129 go c.readLoop()
130 go c.stderrLoop()
131 return c, nil
132}
133
134// NewClientWithContext starts and initializes a client.
135func NewClientWithContext(ctx context.Context, spec ServerConfig, root string, options ClientOptions) (*Client, error) {
136 client, err := NewClientWithOptions(spec, root, options)
137 if err != nil {
138 return nil, err
139 }
140 if err := client.Initialize(ctx); err != nil {
141 _ = client.Close()
142 return nil, err
143 }
144 return client, nil
145}
146
147func (c *Client) Initialize(ctx context.Context) error {
148 if ctx == nil {
149 ctx = context.Background()
150 }
151 folders := []map[string]any{{"uri": mustURI(c.Root), "name": filepath.Base(c.Root)}}
152 params := map[string]any{
153 "processId": os.Getpid(),
154 "rootUri": mustURI(c.Root),
155 "workspaceFolders": folders,
156 "capabilities": map[string]any{
157 "workspace": map[string]any{
158 "applyEdit": true,
159 "workspaceFolders": true,
160 "configuration": true,
161 },
162 "textDocument": map[string]any{
163 "publishDiagnostics": map[string]any{"relatedInformation": true},
164 },
165 },
166 "initializationOptions": c.options.InitializationOptions,
167 "clientInfo": map[string]any{"name": "zut", "version": "lsp"},
168 }
169 result, err := c.Request(ctx, "initialize", params)
170 if err != nil {
171 return fmt.Errorf("initialize %s: %w", c.Spec.ID, err)
172 }
173 var response struct {
174 Capabilities json.RawMessage `json:"capabilities"`
175 }
176 if err := json.Unmarshal(result, &response); err == nil {
177 c.capMu.Lock()
178 c.caps = append(c.caps[:0], response.Capabilities...)
179 c.capMu.Unlock()
180 }
181 if err := c.Notify("initialized", map[string]any{}); err != nil {
182 return err
183 }
184 return nil
185}
186
187// Capabilities returns the initialize result's capabilities.
188func (c *Client) Capabilities() json.RawMessage {
189 c.capMu.RLock()
190 defer c.capMu.RUnlock()
191 return append(json.RawMessage(nil), c.caps...)
192}
193
194// Request sends a raw JSON-RPC request and waits for its response.
195func (c *Client) Request(ctx context.Context, method string, params any) (json.RawMessage, error) {
196 if ctx == nil {
197 ctx = context.Background()
198 }
199 id := atomic.AddInt64(&c.nextID, 1)
200 responseCh := make(chan rpcResponse, 1)
201 c.mu.Lock()
202 if c.closed {
203 c.mu.Unlock()
204 return nil, errors.New("LSP client is closed")
205 }
206 c.pending[id] = responseCh
207 c.mu.Unlock()
208 request := map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}
209 if err := c.send(request); err != nil {
210 c.mu.Lock()
211 delete(c.pending, id)
212 c.mu.Unlock()
213 return nil, err
214 }
215 select {
216 case response := <-responseCh:
217 if response.Error != nil {
218 return nil, fmt.Errorf("LSP %s (%d): %s", method, response.Error.Code, response.Error.Message)
219 }
220 return response.Result, nil
221 case <-ctx.Done():
222 c.mu.Lock()
223 delete(c.pending, id)
224 c.mu.Unlock()
225 _ = c.Notify("$/cancelRequest", map[string]any{"id": id})
226 return nil, ctx.Err()
227 case <-c.done:
228 return nil, c.processError()
229 }
230}
231
232// Notify sends a JSON-RPC notification.
233func (c *Client) Notify(method string, params any) error {
234 return c.send(map[string]any{"jsonrpc": "2.0", "method": method, "params": params})
235}
236
237func (c *Client) send(message any) error {
238 payload, err := json.Marshal(message)
239 if err != nil {
240 return err
241 }
242 c.writeMu.Lock()
243 defer c.writeMu.Unlock()
244 c.mu.Lock()
245 closed := c.closed
246 c.mu.Unlock()
247 if closed {
248 return errors.New("LSP client is closed")
249 }
250 if err := WriteMessage(c.stdin, payload); err != nil {
251 return fmt.Errorf("write LSP message: %w", err)
252 }
253 return nil
254}
255
256// DidOpen publishes a complete document. Callers which already opened the URI
257// should use DidChange for subsequent full-content updates.
258func (c *Client) DidOpen(path, languageID, text string) error {
259 uri, err := pathToURI(path)
260 if err != nil {
261 return err
262 }
263 version, alreadyOpen := c.nextDocumentVersion(uri)
264 if alreadyOpen {
265 return c.Notify("textDocument/didChange", map[string]any{
266 "textDocument": map[string]any{"uri": uri, "version": version},
267 "contentChanges": []map[string]any{{"text": text}},
268 })
269 }
270 return c.Notify("textDocument/didOpen", map[string]any{
271 "textDocument": map[string]any{"uri": uri, "languageId": languageID, "version": version, "text": text},
272 })
273}
274
275// DidChange sends a full-content change, which avoids version/range drift
276// between zut and servers that use different incremental-sync policies.
277func (c *Client) DidChange(path, text string) error {
278 uri, err := pathToURI(path)
279 if err != nil {
280 return err
281 }
282 version, _ := c.nextDocumentVersion(uri)
283 return c.Notify("textDocument/didChange", map[string]any{
284 "textDocument": map[string]any{"uri": uri, "version": version},
285 "contentChanges": []map[string]any{{"text": text}},
286 })
287}
288
289func (c *Client) nextDocumentVersion(uri string) (int, bool) {
290 c.docsMu.Lock()
291 defer c.docsMu.Unlock()
292 version := c.docs[uri]
293 alreadyOpen := version != 0
294 if version == 0 {
295 version = 1
296 } else {
297 version++
298 }
299 c.docs[uri] = version
300 return version, alreadyOpen
301}
302
303func (c *Client) DidSave(path, text string) error {
304 uri, err := pathToURI(path)
305 if err != nil {
306 return err
307 }
308 params := map[string]any{"textDocument": map[string]any{"uri": uri}}
309 if text != "" {
310 params["text"] = text
311 }
312 return c.Notify("textDocument/didSave", params)
313}
314
315func (c *Client) readLoop() {
316 reader := bufio.NewReaderSize(c.stdout, 64*1024)
317 for {
318 payload, err := ReadMessage(reader)
319 if err != nil {
320 if !errors.Is(err, io.EOF) {
321 c.readErr.Store(err.Error())
322 }
323 c.failPending(err)
324 c.signalDone()
325 return
326 }
327 var envelope rpcEnvelope
328 if err := json.Unmarshal(payload, &envelope); err != nil {
329 c.readErr.Store(fmt.Sprintf("decode LSP message: %v", err))
330 continue
331 }
332 if envelope.Method == "" && len(envelope.ID) > 0 {
333 var id int64
334 if json.Unmarshal(envelope.ID, &id) == nil {
335 c.mu.Lock()
336 channel := c.pending[id]
337 delete(c.pending, id)
338 c.mu.Unlock()
339 if channel != nil {
340 channel <- rpcResponse{Result: envelope.Result, Error: envelope.Error}
341 }
342 }
343 continue
344 }
345 if envelope.Method != "" {
346 // Notifications carry ordered state, so handle them on the
347 // reader goroutine. Server requests may block while they wait
348 // for an edit response, so those remain asynchronous.
349 if len(envelope.ID) == 0 || string(envelope.ID) == "null" {
350 c.handleServerMessage(envelope)
351 } else {
352 go c.handleServerMessage(envelope)
353 }
354 }
355 }
356}
357
358func (c *Client) handleServerMessage(message rpcEnvelope) {
359 switch message.Method {
360 case "textDocument/publishDiagnostics":
361 c.handleDiagnostics(message.Params)
362 return
363 case "$/progress":
364 if c.options.OnProgress != nil {
365 c.options.OnProgress(message.Method, message.Params)
366 }
367 return
368 case "window/logMessage", "window/showMessage", "telemetry/event":
369 if c.options.OnLog != nil {
370 var value struct {
371 Message string `json:"message"`
372 }
373 _ = json.Unmarshal(message.Params, &value)
374 c.options.OnLog(value.Message)
375 }
376 if len(message.ID) == 0 || string(message.ID) == "null" {
377 return
378 }
379 _ = c.respond(message.ID, nil, nil)
380 return
381 case "workspace/configuration":
382 _ = c.respond(message.ID, c.configuration(message.Params), nil)
383 case "workspace/workspaceFolders":
384 _ = c.respond(message.ID, []map[string]any{{"uri": mustURI(c.Root), "name": filepath.Base(c.Root)}}, nil)
385 case "workspace/applyEdit":
386 var params struct {
387 Edit WorkspaceEdit `json:"edit"`
388 }
389 if err := json.Unmarshal(message.Params, ¶ms); err != nil {
390 _ = c.respond(message.ID, map[string]any{"applied": false, "failureReason": err.Error()}, nil)
391 return
392 }
393 c.applyEditMu.Lock()
394 err := c.applyEdit(params.Edit)
395 c.applyEditMu.Unlock()
396 result := map[string]any{"applied": err == nil}
397 if err != nil {
398 result["failureReason"] = err.Error()
399 }
400 _ = c.respond(message.ID, result, nil)
401 case "client/registerCapability", "window/workDoneProgress/create":
402 _ = c.respond(message.ID, map[string]any{}, nil)
403 default:
404 if len(message.ID) > 0 && string(message.ID) != "null" {
405 _ = c.respond(message.ID, nil, &rpcError{Code: -32601, Message: "method not supported"})
406 }
407 }
408}
409
410func (c *Client) configuration(params json.RawMessage) []any {
411 var request struct {
412 Items []struct {
413 Section string `json:"section"`
414 } `json:"items"`
415 }
416 _ = json.Unmarshal(params, &request)
417 out := make([]any, len(request.Items))
418 for i, item := range request.Items {
419 if item.Section == "" {
420 out[i] = c.options.Settings
421 continue
422 }
423 out[i] = lookupSetting(c.options.Settings, item.Section)
424 }
425 return out
426}
427
428func lookupSetting(settings map[string]any, section string) any {
429 if settings == nil {
430 return nil
431 }
432 if value, ok := settings[section]; ok {
433 return value
434 }
435 var value any = settings
436 for _, part := range strings.Split(section, ".") {
437 object, ok := value.(map[string]any)
438 if !ok {
439 return nil
440 }
441 value, ok = object[part]
442 if !ok {
443 return nil
444 }
445 }
446 return value
447}
448
449func (c *Client) applyEdit(edit WorkspaceEdit) error {
450 if c.options.ApplyEdit != nil {
451 return c.options.ApplyEdit(edit)
452 }
453 return ApplyWorkspaceEdit(c.Root, edit)
454}
455
456func (c *Client) respond(id json.RawMessage, result any, rpcErr *rpcError) error {
457 message := map[string]any{"jsonrpc": "2.0", "id": id}
458 if rpcErr != nil {
459 message["error"] = rpcErr
460 } else {
461 message["result"] = result
462 }
463 return c.send(message)
464}
465
466func (c *Client) handleDiagnostics(params json.RawMessage) {
467 var value struct {
468 URI string `json:"uri"`
469 Diagnostics []Diagnostic `json:"diagnostics"`
470 }
471 if err := json.Unmarshal(params, &value); err != nil {
472 return
473 }
474 path, err := uriToPath(value.URI)
475 if err != nil {
476 return
477 }
478 for i := range value.Diagnostics {
479 value.Diagnostics[i].Path = path
480 value.Diagnostics[i].Server = c.Spec.ID
481 }
482 if len(value.Diagnostics) == 0 {
483 // Preserve the URI on a clear so Manager can remove only this file's
484 // snapshot rather than accidentally retaining stale diagnostics.
485 value.Diagnostics = []Diagnostic{{Path: path, Server: c.Spec.ID, Clear: true}}
486 }
487 if c.options.OnDiagnostics != nil {
488 c.options.OnDiagnostics(value.Diagnostics)
489 }
490}
491
492func (c *Client) stderrLoop() {
493 if c.stderr == nil {
494 return
495 }
496 var output limitedBuffer
497 output.limit = 1 << 20
498 _, _ = io.Copy(&output, c.stderr)
499 if output.Len() > 0 && c.options.OnLog != nil {
500 c.options.OnLog(strings.TrimSpace(output.String()))
501 }
502}
503
504func (c *Client) failPending(err error) {
505 c.mu.Lock()
506 defer c.mu.Unlock()
507 for id, channel := range c.pending {
508 delete(c.pending, id)
509 channel <- rpcResponse{Error: &rpcError{Code: -32000, Message: err.Error()}}
510 }
511}
512
513func (c *Client) processError() error {
514 if err, ok := c.readErr.Load().(string); ok && err != "" {
515 return fmt.Errorf("LSP process stopped: %s", err)
516 }
517 return errors.New("LSP process stopped")
518}
519
520func (c *Client) signalDone() { c.doneOnce.Do(func() { close(c.done) }) }
521
522// Close stops the process and releases all pipes. It is safe to call more
523// than once and is also used by Manager.Reload.
524func (c *Client) Close() error {
525 var waitErr error
526 c.closeOnce.Do(func() {
527 // Give a cooperative server a short shutdown window before killing
528 // the process. Close remains bounded for servers that do not support
529 // shutdown or are already wedged.
530 if c.cmd.ProcessState == nil {
531 shutdownCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
532 _, _ = c.Request(shutdownCtx, "shutdown", nil)
533 cancel()
534 _ = c.Notify("exit", nil)
535 }
536 // Serialize the closed transition with send so a writer that has
537 // passed its closed check cannot race stdin.Close.
538 c.writeMu.Lock()
539 c.mu.Lock()
540 c.closed = true
541 c.mu.Unlock()
542 c.signalDone()
543 _ = c.stdin.Close()
544 c.writeMu.Unlock()
545 killed := false
546 process := c.cmd.Process
547 exited := make(chan error, 1)
548 go func() { exited <- c.cmd.Wait() }()
549 timer := time.NewTimer(500 * time.Millisecond)
550 select {
551 case waitErr = <-exited:
552 timer.Stop()
553 case <-timer.C:
554 if process != nil {
555 killed = process.Kill() == nil
556 }
557 waitErr = <-exited
558 }
559 if killed {
560 // A process terminated by this cleanup path normally returns an
561 // ExitError; preserve unrelated wait failures such as pipe
562 // errors so callers still get useful lifecycle diagnostics.
563 var exitErr *exec.ExitError
564 if errors.As(waitErr, &exitErr) {
565 waitErr = nil
566 }
567 }
568 _ = c.stdout.Close()
569 _ = c.stderr.Close()
570 c.failPending(errors.New("LSP client closed"))
571 })
572 return waitErr
573}
574
575func mustURI(path string) string {
576 uri, _ := pathToURI(path)
577 return uri
578}
579
580func mergedEnvironment(overrides map[string]string) []string {
581 env := os.Environ()
582 if len(overrides) == 0 {
583 return env
584 }
585 index := make(map[string]int, len(env))
586 for i, item := range env {
587 if key, _, ok := strings.Cut(item, "="); ok {
588 index[key] = i
589 }
590 }
591 for key, value := range overrides {
592 item := key + "=" + value
593 if i, ok := index[key]; ok {
594 env[i] = item
595 } else {
596 env = append(env, item)
597 }
598 }
599 return env
600}
601
602// ResolveCommand searches workspace-local tool directories before PATH.
603func ResolveCommand(cwd, command string) (string, error) {
604 if strings.TrimSpace(command) == "" {
605 return "", errors.New("empty command")
606 }
607 cwd, _ = filepath.Abs(cwd)
608 candidates := []string{}
609 if filepath.IsAbs(command) || strings.ContainsAny(command, `/\\`) {
610 if filepath.IsAbs(command) {
611 candidates = append(candidates, command)
612 } else {
613 candidates = append(candidates, filepath.Join(cwd, command))
614 }
615 } else {
616 // A model often runs in a package subdirectory while the project's
617 // executable lives at the repository root. Search each workspace
618 // ancestor before falling back to PATH.
619 for dir := cwd; dir != ""; dir = filepath.Dir(dir) {
620 for _, local := range []string{filepath.Join("node_modules", ".bin"), filepath.Join(".venv", "bin"), filepath.Join(".venv", "Scripts"), "bin", "Scripts"} {
621 candidates = append(candidates, filepath.Join(dir, local, command))
622 }
623 next := filepath.Dir(dir)
624 if next == dir {
625 break
626 }
627 }
628 }
629 for _, candidate := range candidates {
630 for _, path := range executableVariants(candidate) {
631 if isRunnable(path) {
632 return path, nil
633 }
634 }
635 }
636 if path, err := exec.LookPath(command); err == nil {
637 return path, nil
638 }
639 return "", fmt.Errorf("command %q not found in node_modules/.bin, .venv/bin, bin, or PATH", command)
640}
641
642func executableVariants(path string) []string {
643 if filepath.Ext(path) != "" {
644 return []string{path}
645 }
646 return []string{path, path + ".exe", path + ".cmd", path + ".bat"}
647}
648
649func isRunnable(path string) bool {
650 info, err := os.Stat(path)
651 if err != nil || info.IsDir() {
652 return false
653 }
654 if info.Mode()&0111 != 0 || filepath.Ext(path) == ".cmd" || filepath.Ext(path) == ".bat" || filepath.Ext(path) == ".exe" {
655 return true
656 }
657 return false
658}
659
660// WaitForDiagnostics gives asynchronous LSP servers a bounded opportunity to
661// publish diagnostics after didOpen/didChange.
662func (c *Client) WaitForDiagnostics(ctx context.Context, delay time.Duration) error {
663 if ctx == nil {
664 ctx = context.Background()
665 }
666 if delay <= 0 {
667 return nil
668 }
669 timer := time.NewTimer(delay)
670 defer timer.Stop()
671 select {
672 case <-timer.C:
673 return nil
674 case <-ctx.Done():
675 return ctx.Err()
676 case <-c.done:
677 return c.processError()
678 }
679}