git.iko.soy/crush
on main help sign in
Merge remote-tracking branch 'origin/main' into fast-ui
e986e5dChristian Rocha
.github/cla-signatures.json+8
compare with main
unifiedsplit
@@ -1751,6 +1751,14 @@
1751 "created_at": "2026-05-12T22:25:09Z",
1751 "created_at": "2026-05-12T22:25:09Z",
1752 "repoId": 987670088,
1752 "repoId": 987670088,
1753 "pullRequestNo": 2892
1753 "pullRequestNo": 2892
1754+ },
1755+ {
1756+ "name": "akhenakh",
1757+ "id": 218430,
1758+ "comment_id": 4446905473,
1759+ "created_at": "2026-05-14T02:28:13Z",
1760+ "repoId": 987670088,
1761+ "pullRequestNo": 2912
1754 }
1762 }
1755 ]
1763 ]
1756}
1764}
internal/client/client.go+8
@@ -121,6 +121,14 @@ func (c *Client) ShutdownServer(ctx context.Context) error {
121 return nil
121 return nil
122}
122}
123
123
124+// Dial opens a connection to the server using the same scheme-aware
125+// logic the client uses for its HTTP transport. Exposed so callers can
126+// reuse the dialer when they need to construct sibling HTTP transports
127+// (e.g. a readiness probe in the CLI).
128+func (c *Client) Dial(ctx context.Context, network, address string) (net.Conn, error) {
129+ return c.dialer(ctx, network, address)
130+}
131+
124func (c *Client) dialer(ctx context.Context, network, address string) (net.Conn, error) {
132func (c *Client) dialer(ctx context.Context, network, address string) (net.Conn, error) {
125 d := net.Dialer{
133 d := net.Dialer{
126 Timeout: 30 * time.Second,
134 Timeout: 30 * time.Second,
internal/cmd/clientserverrace/race_test.gonew file+361
@@ -0,0 +1,361 @@
1+// Package clientserverrace_test is a regression test for the
2+// CRUSH_CLIENT_SERVER=1 socket-init race documented in
3+// docs/notes/2026-05-11-client-server-socket-init-race.md (item F5).
4+//
5+// It lives in its own directory so it can build even if other test
6+// files in internal/cmd are temporarily broken — this test only needs
7+// the binary, not the cmd package.
8+package clientserverrace_test
9+
10+import (
11+ "context"
12+ "fmt"
13+ "net"
14+ "net/http"
15+ "os"
16+ "os/exec"
17+ "path/filepath"
18+ "runtime"
19+ "strings"
20+ "sync"
21+ "sync/atomic"
22+ "testing"
23+ "time"
24+)
25+
26+// readinessErrSubstr is the user-visible error string emitted by
27+// ensureServer when it gives up waiting for the server socket /
28+// readiness probe (internal/cmd/root.go). Seeing this in any client's
29+// output means the race fired.
30+const readinessErrSubstr = "failed to initialize crush server"
31+
32+// numClients is intentionally larger than the typical CPU count to
33+// ensure the spawn lock + readiness probe are exercised under
34+// contention.
35+const numClients = 8
36+
37+// clientTimeout bounds each child invocation. It only needs to be long
38+// enough for the spawn-and-readiness phase to complete on a cold cache;
39+// after that, the client may legitimately keep running (e.g.
40+// subscribing to server events) and we'll cancel it. The race we care
41+// about is observable strictly within ensureServer.
42+const clientTimeout = 15 * time.Second
43+
44+func TestClientServerSpawnRace(t *testing.T) {
45+ if testing.Short() {
46+ t.Skip("skipping client/server spawn race test in -short mode")
47+ }
48+ // The race and its fix are unix-socket specific. Windows uses
49+ // named pipes via a different code path; not covered here.
50+ if runtime.GOOS == "windows" {
51+ t.Skip("skipping unix-socket specific race test on windows")
52+ }
53+ if _, err := exec.LookPath("go"); err != nil {
54+ t.Skip("skipping: 'go' not available on PATH")
55+ }
56+
57+ repoRoot := repoRootFromTest(t)
58+ bin := buildCrushBinary(t, repoRoot)
59+
60+ // Use /tmp directly so the unix socket path stays under the
61+ // 104-char sockaddr_un limit on darwin. t.TempDir() can return a
62+ // path inside /var/folders/... that is too long.
63+ runDir, err := os.MkdirTemp("/tmp", "crush-race-")
64+ if err != nil {
65+ t.Fatalf("mkdtemp: %v", err)
66+ }
67+ t.Cleanup(func() { _ = os.RemoveAll(runDir) })
68+
69+ socketPath := filepath.Join(runDir, "crush.sock")
70+ host := "unix://" + socketPath
71+
72+ // Fresh, isolated XDG/HOME so we don't touch the user's real
73+ // state or any other test's cache. These all live under runDir
74+ // so cleanup is one RemoveAll.
75+ cacheHome := filepath.Join(runDir, "cache")
76+ dataHome := filepath.Join(runDir, "data")
77+ configHome := filepath.Join(runDir, "config")
78+ homeDir := filepath.Join(runDir, "home")
79+ for _, d := range []string{cacheHome, dataHome, configHome, homeDir} {
80+ if err := os.MkdirAll(d, 0o700); err != nil {
81+ t.Fatalf("mkdir %s: %v", d, err)
82+ }
83+ }
84+
85+ env := append(
86+ os.Environ(),
87+ "CRUSH_CLIENT_SERVER=1",
88+ "XDG_CACHE_HOME="+cacheHome,
89+ "XDG_DATA_HOME="+dataHome,
90+ "XDG_CONFIG_HOME="+configHome,
91+ "HOME="+homeDir,
92+ // Belt-and-suspenders: if anything tries to talk to a real
93+ // provider, fail loudly rather than make a network call.
94+ "CRUSH_DISABLE_PROVIDER_AUTO_UPDATE=1",
95+ )
96+
97+ // Make sure no server is up before we start.
98+ if _, err := os.Stat(socketPath); err == nil {
99+ t.Fatalf("socket %s exists before test started", socketPath)
100+ }
101+
102+ // Always try to shut down any server we spawned, regardless of
103+ // outcome.
104+ t.Cleanup(func() { shutdownServer(t, socketPath) })
105+
106+ type result struct {
107+ idx int
108+ stdout string
109+ stderr string
110+ }
111+ results := make(chan result, numClients)
112+
113+ // Probe /v1/health concurrently while the clients are still
114+ // running. The server self-shuts-down when the last workspace is
115+ // released (internal/backend/backend.go:DeleteWorkspace), so once
116+ // all clients exit cleanly the socket may legitimately be gone —
117+ // asserting the socket post-hoc would race with that documented
118+ // self-shutdown. Instead we require that during the parallel run
119+ // at least one /v1/health probe got a 2xx, which proves the
120+ // spawn-and-readiness path actually produced a live server.
121+ var sawHealthy atomic.Bool
122+ probeDone := make(chan struct{})
123+ stopProbe := make(chan struct{})
124+
125+ var wg sync.WaitGroup
126+ start := make(chan struct{})
127+
128+ go func() {
129+ defer close(probeDone)
130+ <-start
131+ deadline := time.Now().Add(clientTimeout)
132+ for time.Now().Before(deadline) {
133+ select {
134+ case <-stopProbe:
135+ return
136+ default:
137+ }
138+ if err := pingHealth(socketPath); err == nil {
139+ sawHealthy.Store(true)
140+ return
141+ }
142+ select {
143+ case <-stopProbe:
144+ return
145+ case <-time.After(50 * time.Millisecond):
146+ }
147+ }
148+ }()
149+
150+ for i := range numClients {
151+ wg.Add(1)
152+ go func(i int) {
153+ defer wg.Done()
154+
155+ // Each client gets its own working directory so the
156+ // per-client workspace registration paths don't collide
157+ // in confusing ways.
158+ cwd := filepath.Join(runDir, fmt.Sprintf("ws-%d", i))
159+ if err := os.MkdirAll(cwd, 0o700); err != nil {
160+ results <- result{idx: i, stderr: fmt.Sprintf("mkdir cwd: %v", err)}
161+ return
162+ }
163+
164+ ctx, cancel := context.WithTimeout(context.Background(), clientTimeout)
165+ defer cancel()
166+
167+ // `crush run` exercises connectToServer (which is where
168+ // the readiness race lives). On a fresh sandbox the
169+ // command may legitimately keep running past the race
170+ // (e.g. waiting on event subscriptions); the context
171+ // timeout above bounds that. We assert race outcomes
172+ // purely from output, not exit code.
173+ c := exec.CommandContext(
174+ ctx, bin,
175+ "--host", host,
176+ "--cwd", cwd,
177+ "run", "hi",
178+ )
179+ c.Env = env
180+ var outBuf, errBuf strings.Builder
181+ c.Stdout = &outBuf
182+ c.Stderr = &errBuf
183+
184+ <-start
185+ _ = c.Run()
186+ results <- result{
187+ idx: i,
188+ stdout: outBuf.String(),
189+ stderr: errBuf.String(),
190+ }
191+ }(i)
192+ }
193+
194+ close(start) // release all clients as simultaneously as possible
195+ wg.Wait()
196+ close(results)
197+ close(stopProbe)
198+ <-probeDone
199+
200+ var raceFailures []string
201+ for r := range results {
202+ if strings.Contains(r.stderr, readinessErrSubstr) ||
203+ strings.Contains(r.stdout, readinessErrSubstr) {
204+ raceFailures = append(raceFailures, fmt.Sprintf(
205+ "client %d: readiness error in output\nstderr:\n%s\nstdout:\n%s",
206+ r.idx, r.stderr, r.stdout,
207+ ))
208+ }
209+ }
210+
211+ if len(raceFailures) > 0 {
212+ t.Fatalf(
213+ "client/server spawn race regressed: %d/%d clients failed\n\n%s",
214+ len(raceFailures), numClients,
215+ strings.Join(raceFailures, "\n---\n"),
216+ )
217+ }
218+
219+ // Positive sanity check: at some point during the parallel run a
220+ // /v1/health probe must have succeeded. We deliberately do *not*
221+ // stat the socket post-hoc: when every client returns cleanly
222+ // (e.g. exits early because no providers are configured), the
223+ // last DeleteWorkspace triggers the server's self-shutdown and
224+ // the socket disappears. That is correct behaviour, not a race
225+ // regression.
226+ if !sawHealthy.Load() {
227+ t.Fatalf("no /v1/health probe succeeded on %s while %d clients were running",
228+ socketPath, numClients)
229+ }
230+}
231+
232+// pingHealth issues a single GET /v1/health over the unix socket and
233+// requires a 2xx response.
234+func pingHealth(socketPath string) error {
235+ tr := &http.Transport{
236+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
237+ var d net.Dialer
238+ return d.DialContext(ctx, "unix", socketPath)
239+ },
240+ }
241+ defer tr.CloseIdleConnections()
242+ hc := &http.Client{Transport: tr, Timeout: 2 * time.Second}
243+
244+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
245+ defer cancel()
246+ req, err := http.NewRequestWithContext(ctx, http.MethodGet,
247+ "http://crush.local/v1/health", nil)
248+ if err != nil {
249+ return err
250+ }
251+ rsp, err := hc.Do(req)
252+ if err != nil {
253+ return err
254+ }
255+ defer rsp.Body.Close()
256+ if rsp.StatusCode < 200 || rsp.StatusCode >= 300 {
257+ return fmt.Errorf("health check returned %s", rsp.Status)
258+ }
259+ return nil
260+}
261+
262+// repoRootFromTest walks up from this test file's directory to find
263+// the repo root (the directory containing go.mod). Walking up by a
264+// fixed count is fragile across reorganisations.
265+func repoRootFromTest(t *testing.T) string {
266+ t.Helper()
267+ cwd, err := os.Getwd()
268+ if err != nil {
269+ t.Fatalf("getwd: %v", err)
270+ }
271+ dir := cwd
272+ for {
273+ if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
274+ return dir
275+ }
276+ parent := filepath.Dir(dir)
277+ if parent == dir {
278+ t.Fatalf("could not find go.mod walking up from %s", cwd)
279+ }
280+ dir = parent
281+ }
282+}
283+
284+// buildCrushBinary builds the crush binary once at the start of the
285+// test and returns the absolute path. Subsequent t.Cleanup removes
286+// the built artefact.
287+func buildCrushBinary(t *testing.T, repoRoot string) string {
288+ t.Helper()
289+
290+ binDir, err := os.MkdirTemp("", "crush-race-bin-")
291+ if err != nil {
292+ t.Fatalf("mkdtemp bin: %v", err)
293+ }
294+ t.Cleanup(func() { _ = os.RemoveAll(binDir) })
295+
296+ binPath := filepath.Join(binDir, "crush")
297+
298+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
299+ defer cancel()
300+
301+ cmd := exec.CommandContext(ctx, "go", "build", "-o", binPath, ".")
302+ cmd.Dir = repoRoot
303+ // Match the project's standard build flags. CGO_ENABLED=0 keeps
304+ // the binary statically linked and avoids surprising the test on
305+ // hosts without a C toolchain.
306+ cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
307+ out, err := cmd.CombinedOutput()
308+ if err != nil {
309+ t.Fatalf("go build crush: %v\n%s", err, out)
310+ }
311+ return binPath
312+}
313+
314+// shutdownServer best-effort terminates any crush server bound to
315+// socketPath by POSTing to /v1/control. We don't import the project's
316+// own client package to keep this test free of internal API churn.
317+func shutdownServer(t *testing.T, socketPath string) {
318+ t.Helper()
319+ if _, err := os.Stat(socketPath); err != nil {
320+ return
321+ }
322+
323+ tr := &http.Transport{
324+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
325+ var d net.Dialer
326+ return d.DialContext(ctx, "unix", socketPath)
327+ },
328+ }
329+ hc := &http.Client{Transport: tr, Timeout: 5 * time.Second}
330+ defer tr.CloseIdleConnections()
331+
332+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
333+ defer cancel()
334+
335+ body := strings.NewReader(`{"command":"shutdown"}`)
336+ req, err := http.NewRequestWithContext(ctx, http.MethodPost,
337+ "http://crush.local/v1/control", body)
338+ if err != nil {
339+ t.Logf("shutdown: build request: %v", err)
340+ return
341+ }
342+ req.Header.Set("Content-Type", "application/json")
343+
344+ resp, err := hc.Do(req)
345+ if err != nil {
346+ // Server may already be gone — not an error.
347+ t.Logf("shutdown: %v (probably already exited)", err)
348+ return
349+ }
350+ _ = resp.Body.Close()
351+
352+ // Wait briefly for the socket to disappear so the next test
353+ // using the same path doesn't race.
354+ deadline := time.Now().Add(5 * time.Second)
355+ for time.Now().Before(deadline) {
356+ if _, err := os.Stat(socketPath); err != nil {
357+ return
358+ }
359+ time.Sleep(50 * time.Millisecond)
360+ }
361+}
internal/cmd/root.go+221−47
@@ -9,6 +9,8 @@ import (
9 "io"
9 "io"
10 "io/fs"
10 "io/fs"
11 "log/slog"
11 "log/slog"
12+ "net"
13+ "net/http"
12 "net/url"
14 "net/url"
13 "os"
15 "os"
14 "os/exec"
16 "os/exec"
@@ -357,22 +359,7 @@ func connectToServer(cmd *cobra.Command) (*client.Client, *proto.Workspace, func
357
359
358 ws, err := c.CreateWorkspace(ctx, wsReq)
360 ws, err := c.CreateWorkspace(ctx, wsReq)
359 if err != nil {
361 if err != nil {
360− // The server socket may exist before the HTTP handler is ready.
361− // Retry a few times with a short backoff.
362− for range 5 {
363− select {
364− case <-ctx.Done():
365− return nil, nil, nil, ctx.Err()
366− case <-time.After(200 * time.Millisecond):
367− }
368− ws, err = c.CreateWorkspace(ctx, wsReq)
369− if err == nil {
370− break
371− }
372− }
373− if err != nil {
374− return nil, nil, nil, fmt.Errorf("failed to create workspace: %v", err)
362+ return nil, nil, nil, fmt.Errorf("failed to create workspace: %v", err)
375− }
376 }
363 }
377
364
378 if shouldEnableMetrics(ws.Config) {
365 if shouldEnableMetrics(ws.Config) {
@@ -396,55 +383,239 @@ func ensureServer(cmd *cobra.Command, hostURL *url.URL) error {
396 switch hostURL.Scheme {
383 switch hostURL.Scheme {
397 case "unix", "npipe":
384 case "unix", "npipe":
398 needsStart := false
385 needsStart := false
386+ _, statErr := os.Stat(hostURL.Host)
387+ switch {
388+ case statErr == nil:
389+ restarted, err := restartIfStale(cmd, hostURL)
390+ if err != nil {
391+ slog.Warn("Failed to check server version", "error", err)
392+ }
393+ needsStart = restarted || err != nil
399− if _, err := os.Stat(hostURL.Host); err != nil && errors.Is(err, fs.ErrNotExist) {
394+ case errors.Is(statErr, fs.ErrNotExist):
400 needsStart = true
395 needsStart = true
401− } else if err == nil {
396+ default:
397+ slog.Warn("Unexpected error stat'ing server socket, attempting cleanup",
398+ "path", hostURL.Host, "error", statErr)
402− if err := restartIfStale(cmd, hostURL); err != nil {
399+ if err := os.Remove(hostURL.Host); err != nil && !errors.Is(err, fs.ErrNotExist) {
403− slog.Warn("Failed to check server version, restarting", "error", err)
400+ return fmt.Errorf("failed to remove stale server socket %q: %v", hostURL.Host, err)
404− needsStart = true
405 }
401 }
402+ needsStart = true
406 }
403 }
407
404
408 if needsStart {
405 if needsStart {
409− if err := startDetachedServer(cmd); err != nil {
406+ if err := spawnAndWaitReady(cmd, hostURL); err != nil {
410− return err
407+ return fmt.Errorf("failed to initialize crush server: %v", err)
411 }
408 }
409+ return nil
412 }
410 }
413
411
414− var err error
412+ if err := waitForServerReady(cmd.Context(), hostURL); err != nil {
415− for range 10 {
413+ return fmt.Errorf("failed to initialize crush server: %v", err)
416− _, err = os.Stat(hostURL.Host)
414+ }
417− if err == nil {
415+ }
418− break
416+
417+ return nil
418+}
419+
420+// spawnAndWaitReady serializes the spawn-and-wait-for-readiness sequence
421+// across concurrent clients via an exclusive flock on
422+// $XDG_CACHE_HOME/crush/server-<safeHost>/start.lock.
423+//
424+// After acquiring the lock it re-probes readiness so that a client that
425+// blocked while another client was spawning can skip its own spawn and
426+// just use the now-running server. The lock is held only for the
427+// duration of "spawn + readiness probe" and released before the caller
428+// resumes its normal lifetime.
429+func spawnAndWaitReady(cmd *cobra.Command, hostURL *url.URL) error {
430+ chDir, err := perHostServerDir(hostURL)
431+ if err != nil {
432+ return err
433+ }
434+ release, err := acquireSpawnLock(filepath.Join(chDir, "start.lock"))
435+ if err != nil {
436+ // If the lock itself is unavailable, fall back to the
437+ // unsynchronized path rather than blocking the user.
438+ slog.Warn("Failed to acquire spawn lock, proceeding without single-flight", "error", err)
439+ if err := startDetachedServer(cmd, hostURL); err != nil {
440+ return err
441+ }
442+ return waitForServerReady(cmd.Context(), hostURL)
443+ }
444+ defer release()
445+
446+ // Another client may have just finished spawning while we were
447+ // waiting on the lock; if the server is already responsive, skip
448+ // the spawn entirely.
449+ probeCtx, cancel := context.WithTimeout(cmd.Context(), 200*time.Millisecond)
450+ probeErr := quickHealthProbe(probeCtx, hostURL)
451+ cancel()
452+ if probeErr == nil {
453+ return nil
454+ }
455+
456+ if err := startDetachedServer(cmd, hostURL); err != nil {
457+ return err
458+ }
459+ return waitForServerReady(cmd.Context(), hostURL)
460+}
461+
462+// quickHealthProbe issues a single readiness request with the caller's
463+// context and returns nil iff the server is responsive right now.
464+func quickHealthProbe(ctx context.Context, hostURL *url.URL) error {
465+ httpClient, reqURL, err := readinessHTTPClient(hostURL)
466+ if err != nil {
467+ return err
468+ }
469+ return probeHealth(ctx, httpClient, reqURL, hostURL)
470+}
471+
472+// perHostServerDir returns (and creates) the cache directory used for
473+// per-host server state (logs, start.lock, etc.). The path is derived
474+// from the parsed host URL rather than the global flag so the same key
475+// is computed regardless of where the host came from.
476+func perHostServerDir(hostURL *url.URL) (string, error) {
477+ chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeHostName(hostURL))
478+ if err := os.MkdirAll(chDir, 0o700); err != nil {
479+ return "", fmt.Errorf("failed to create server working directory: %v", err)
480+ }
481+ return chDir, nil
482+}
483+
484+// safeHostName returns a filesystem-safe identifier for hostURL,
485+// suitable for use as a directory name. It mirrors the input shape of
486+// the --host flag so client and server compute the same key.
487+func safeHostName(hostURL *url.URL) string {
488+ return safeNameRegexp.ReplaceAllString(
489+ hostURL.Scheme+"://"+hostURL.Host+hostURL.Path, "_")
490+}
491+
492+// serverReadyTimeout returns the total budget for the readiness probe.
493+// Overridable via CRUSH_SERVER_READY_TIMEOUT (parsed as a Go duration).
494+func serverReadyTimeout() time.Duration {
495+ const def = 10 * time.Second
496+ v := os.Getenv("CRUSH_SERVER_READY_TIMEOUT")
497+ if v == "" {
498+ return def
499+ }
500+ d, err := time.ParseDuration(v)
501+ if err != nil || d <= 0 {
502+ return def
503+ }
504+ return d
419− }
505+}
420− select {
506+
421− case <-cmd.Context().Done():
507+// waitForServerReady polls GET /v1/health until the server responds with
508+// any 2xx status or the total timeout elapses. Each attempt uses a short
509+// per-attempt timeout so a hung listener doesn't burn the whole budget.
510+//
511+// The HTTP transport is built to mirror how *client.Client dials so the
512+// same unix socket / npipe / tcp setups all work uniformly here.
513+func waitForServerReady(ctx context.Context, hostURL *url.URL) error {
514+ httpClient, reqURL, err := readinessHTTPClient(hostURL)
515+ if err != nil {
422− return cmd.Context().Err()
516+ return err
423− case <-time.After(100 * time.Millisecond):
517+ }
518+
519+ const perAttempt = 100 * time.Millisecond
520+ deadline := time.Now().Add(serverReadyTimeout())
521+
522+ var lastErr error
523+ for {
524+ if err := ctx.Err(); err != nil {
525+ return err
526+ }
527+ if time.Now().After(deadline) {
528+ if lastErr != nil {
529+ return lastErr
424 }
530 }
531+ return fmt.Errorf("timed out waiting for server readiness")
425 }
532 }
533+
534+ attemptCtx, cancel := context.WithTimeout(ctx, perAttempt)
535+ err := probeHealth(attemptCtx, httpClient, reqURL, hostURL)
536+ cancel()
426− if err != nil {
537+ if err == nil {
427− return fmt.Errorf("failed to initialize crush server: %v", err)
538+ return nil
539+ }
540+ lastErr = err
541+
542+ select {
543+ case <-ctx.Done():
544+ return ctx.Err()
545+ case <-time.After(perAttempt):
428 }
546 }
429 }
547 }
548+}
430
549
550+// readinessHTTPClient builds an *http.Client whose transport dials the
551+// server using the same scheme-aware logic as *client.Client (unix
552+// socket, named pipe, or tcp).
553+func readinessHTTPClient(hostURL *url.URL) (*http.Client, string, error) {
554+ c, err := client.NewClient("", hostURL.Scheme, hostURL.Host)
555+ if err != nil {
556+ return nil, "", err
557+ }
558+
559+ tr := http.DefaultTransport.(*http.Transport).Clone()
560+ tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
561+ return c.Dial(ctx, network, addr)
562+ }
563+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
564+ tr.DisableCompression = true
565+ }
566+
567+ httpClient := &http.Client{Transport: tr}
568+
569+ // For unix sockets / named pipes we still need a syntactically valid
570+ // HTTP URL; the actual address is resolved by the dialer.
571+ host := hostURL.Host
572+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
573+ host = client.DummyHost
574+ }
575+ reqURL := (&url.URL{Scheme: "http", Host: host, Path: "/v1/health"}).String()
576+ return httpClient, reqURL, nil
577+}
578+
579+// probeHealth issues a single GET to the readiness endpoint and treats
580+// any 2xx response as success.
581+func probeHealth(ctx context.Context, h *http.Client, reqURL string, hostURL *url.URL) error {
582+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
583+ if err != nil {
584+ return err
585+ }
586+ if hostURL.Scheme == "unix" || hostURL.Scheme == "npipe" {
587+ req.Host = client.DummyHost
588+ }
589+ rsp, err := h.Do(req)
590+ if err != nil {
591+ return err
592+ }
593+ defer rsp.Body.Close()
594+ _, _ = io.Copy(io.Discard, rsp.Body)
595+ if rsp.StatusCode < 200 || rsp.StatusCode >= 300 {
596+ return fmt.Errorf("server health check failed: %s", rsp.Status)
597+ }
431 return nil
598 return nil
432}
599}
433
600
434// restartIfStale checks whether the running server matches the current
601// restartIfStale checks whether the running server matches the current
435// client version. When they differ, it sends a shutdown command and
602// client version. When they differ, it sends a shutdown command and
436// removes the stale socket so the caller can start a fresh server.
603// removes the stale socket so the caller can start a fresh server.
604+//
605+// It returns restarted=true when it has shut down a stale server and the
606+// caller must spawn a new one. When the server matches the client version
607+// (or the check itself fails), restarted is false.
437−func restartIfStale(cmd *cobra.Command, hostURL *url.URL) error {
608+func restartIfStale(cmd *cobra.Command, hostURL *url.URL) (restarted bool, err error) {
438 c, err := client.NewClient("", hostURL.Scheme, hostURL.Host)
609 c, err := client.NewClient("", hostURL.Scheme, hostURL.Host)
439 if err != nil {
610 if err != nil {
440− return err
611+ return false, err
441 }
612 }
442 vi, err := c.VersionInfo(cmd.Context())
613 vi, err := c.VersionInfo(cmd.Context())
443 if err != nil {
614 if err != nil {
444− return err
615+ return false, err
445 }
616 }
446 if vi.Version == version.Version {
617 if vi.Version == version.Version {
447− return nil
618+ return false, nil
448 }
619 }
449 slog.Info("Server version mismatch, restarting",
620 slog.Info("Server version mismatch, restarting",
450 "server", vi.Version,
621 "server", vi.Version,
@@ -458,27 +629,26 @@ func restartIfStale(cmd *cobra.Command, hostURL *url.URL) error {
458 }
629 }
459 select {
630 select {
460 case <-cmd.Context().Done():
631 case <-cmd.Context().Done():
461− return cmd.Context().Err()
632+ return true, cmd.Context().Err()
462 case <-time.After(100 * time.Millisecond):
633 case <-time.After(100 * time.Millisecond):
463 }
634 }
464 }
635 }
465 // Force-remove if the socket is still lingering.
636 // Force-remove if the socket is still lingering.
466 _ = os.Remove(hostURL.Host)
637 _ = os.Remove(hostURL.Host)
467− return nil
638+ return true, nil
468}
639}
469
640
470var safeNameRegexp = regexp.MustCompile(`[^a-zA-Z0-9._-]`)
641var safeNameRegexp = regexp.MustCompile(`[^a-zA-Z0-9._-]`)
471
642
472−func startDetachedServer(cmd *cobra.Command) error {
643+func startDetachedServer(cmd *cobra.Command, hostURL *url.URL) error {
473 exe, err := os.Executable()
644 exe, err := os.Executable()
474 if err != nil {
645 if err != nil {
475 return fmt.Errorf("failed to get executable path: %v", err)
646 return fmt.Errorf("failed to get executable path: %v", err)
476 }
647 }
477
648
478− safeClientHost := safeNameRegexp.ReplaceAllString(clientHost, "_")
649+ chDir, err := perHostServerDir(hostURL)
479− chDir := filepath.Join(config.GlobalCacheDir(), "server-"+safeClientHost)
480− if err := os.MkdirAll(chDir, 0o700); err != nil {
650+ if err != nil {
481− return fmt.Errorf("failed to create server working directory: %v", err)
651+ return err
482 }
652 }
483
653
484 cmdArgs := []string{"server"}
654 cmdArgs := []string{"server"}
@@ -486,7 +656,11 @@ func startDetachedServer(cmd *cobra.Command) error {
486 cmdArgs = append(cmdArgs, "--host", clientHost)
656 cmdArgs = append(cmdArgs, "--host", clientHost)
487 }
657 }
488
658
659+ // Use context.Background() so the parent's context cancellation does not
660+ // kill the spawned server. detachProcess (Setsid on !windows,
661+ // DETACHED_PROCESS on windows) is what truly detaches the child from
662+ // this process's lifetime.
489− c := exec.CommandContext(cmd.Context(), exe, cmdArgs...)
663+ c := exec.CommandContext(context.Background(), exe, cmdArgs...)
490 stdoutPath := filepath.Join(chDir, "stdout.log")
664 stdoutPath := filepath.Join(chDir, "stdout.log")
491 stderrPath := filepath.Join(chDir, "stderr.log")
665 stderrPath := filepath.Join(chDir, "stderr.log")
492 detachProcess(c)
666 detachProcess(c)
internal/cmd/server.go+6−6
@@ -42,7 +42,12 @@ var serverCmd = &cobra.Command{
42 return fmt.Errorf("failed to load configuration: %v", err)
42 return fmt.Errorf("failed to load configuration: %v", err)
43 }
43 }
44
44
45+ hostURL, err := server.ParseHostURL(serverHost)
46+ if err != nil {
47+ return fmt.Errorf("invalid server host: %v", err)
48+ }
49+
45− logFile := filepath.Join(config.GlobalCacheDir(), "server-"+safeNameRegexp.ReplaceAllString(serverHost, "_"), "crush.log")
50+ logFile := filepath.Join(config.GlobalCacheDir(), "server-"+safeHostName(hostURL), "crush.log")
46
51
47 if term.IsTerminal(os.Stderr.Fd()) {
52 if term.IsTerminal(os.Stderr.Fd()) {
48 crushlog.Setup(logFile, debug, os.Stderr)
53 crushlog.Setup(logFile, debug, os.Stderr)
@@ -50,11 +55,6 @@ var serverCmd = &cobra.Command{
50 crushlog.Setup(logFile, debug)
55 crushlog.Setup(logFile, debug)
51 }
56 }
52
57
53− hostURL, err := server.ParseHostURL(serverHost)
54− if err != nil {
55− return fmt.Errorf("invalid server host: %v", err)
56− }
57−
58 srv := server.NewServer(cfg, hostURL.Scheme, hostURL.Host)
58 srv := server.NewServer(cfg, hostURL.Scheme, hostURL.Host)
59 srv.SetLogger(slog.Default())
59 srv.SetLogger(slog.Default())
60 slog.Info("Starting Crush server...", "addr", serverHost)
60 slog.Info("Starting Crush server...", "addr", serverHost)
internal/cmd/spawnlock_other.gonew file+28
@@ -0,0 +1,28 @@
1+//go:build !windows
2+
3+package cmd
4+
5+import (
6+ "fmt"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
10+)
11+
12+// acquireSpawnLock takes an exclusive flock on the given file (creating
13+// it if necessary) and returns a release function that unlocks and
14+// closes the file. Blocks until the lock is acquired.
15+func acquireSpawnLock(path string) (func(), error) {
16+ f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
17+ if err != nil {
18+ return nil, fmt.Errorf("open spawn lock %q: %v", path, err)
19+ }
20+ if err := unix.Flock(int(f.Fd()), unix.LOCK_EX); err != nil {
21+ _ = f.Close()
22+ return nil, fmt.Errorf("flock spawn lock %q: %v", path, err)
23+ }
24+ return func() {
25+ _ = unix.Flock(int(f.Fd()), unix.LOCK_UN)
26+ _ = f.Close()
27+ }, nil
28+}
internal/cmd/spawnlock_windows.gonew file+32
@@ -0,0 +1,32 @@
1+//go:build windows
2+
3+package cmd
4+
5+import (
6+ "fmt"
7+ "math"
8+ "os"
9+
10+ "golang.org/x/sys/windows"
11+)
12+
13+// acquireSpawnLock takes an exclusive lock on the given file (creating
14+// it if necessary) using LockFileEx, and returns a release function
15+// that unlocks and closes the file. Blocks until the lock is acquired.
16+func acquireSpawnLock(path string) (func(), error) {
17+ f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
18+ if err != nil {
19+ return nil, fmt.Errorf("open spawn lock %q: %v", path, err)
20+ }
21+ h := windows.Handle(f.Fd())
22+ ol := new(windows.Overlapped)
23+ if err := windows.LockFileEx(h, windows.LOCKFILE_EXCLUSIVE_LOCK, 0, math.MaxUint32, math.MaxUint32, ol); err != nil {
24+ _ = f.Close()
25+ return nil, fmt.Errorf("LockFileEx spawn lock %q: %v", path, err)
26+ }
27+ return func() {
28+ ol := new(windows.Overlapped)
29+ _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, math.MaxUint32, math.MaxUint32, ol)
30+ _ = f.Close()
31+ }, nil
32+}
internal/proto/message.go+2
@@ -153,6 +153,8 @@ type ToolResult struct {
153 ToolCallID string `json:"tool_call_id"`
153 ToolCallID string `json:"tool_call_id"`
154 Name string `json:"name"`
154 Name string `json:"name"`
155 Content string `json:"content"`
155 Content string `json:"content"`
156+ Data string `json:"data,omitempty"`
157+ MIMEType string `json:"mime_type,omitempty"`
156 Metadata string `json:"metadata"`
158 Metadata string `json:"metadata"`
157 IsError bool `json:"is_error"`
159 IsError bool `json:"is_error"`
158}
160}
internal/proto/permission.go+6
@@ -118,6 +118,12 @@ func unmarshalToolParams(toolName string, raw json.RawMessage) (any, error) {
118 return nil, err
118 return nil, err
119 }
119 }
120 return params, nil
120 return params, nil
121+ case AgenticFetchToolName:
122+ var params AgenticFetchPermissionsParams
123+ if err := json.Unmarshal(raw, &params); err != nil {
124+ return nil, err
125+ }
126+ return params, nil
121 case ViewToolName:
127 case ViewToolName:
122 var params ViewPermissionsParams
128 var params ViewPermissionsParams
123 if err := json.Unmarshal(raw, &params); err != nil {
129 if err := json.Unmarshal(raw, &params); err != nil {
internal/proto/permission_test.gonew file+186
@@ -0,0 +1,186 @@
1+package proto_test
2+
3+import (
4+ "encoding/json"
5+ "testing"
6+
7+ "github.com/charmbracelet/crush/internal/agent/tools"
8+ "github.com/charmbracelet/crush/internal/proto"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+// TestPermissionRequestParamsTypeAssertable guards the permission
13+// dialog's type assertions across the client/server boundary. The TUI
14+// asserts PermissionRequest.Params to tools.*PermissionsParams; when
15+// the request round-trips over the SSE wire (server → client), the
16+// decoded value must be the same Go type, otherwise the dialog
17+// renders empty content.
18+func TestPermissionRequestParamsTypeAssertable(t *testing.T) {
19+ t.Parallel()
20+
21+ tests := []struct {
22+ name string
23+ toolName string
24+ params any
25+ assert func(t *testing.T, got any)
26+ }{
27+ {
28+ name: "bash",
29+ toolName: tools.BashToolName,
30+ params: tools.BashPermissionsParams{
31+ Description: "list files",
32+ Command: "ls -la",
33+ WorkingDir: "/tmp",
34+ RunInBackground: false,
35+ },
36+ assert: func(t *testing.T, got any) {
37+ v, ok := got.(tools.BashPermissionsParams)
38+ require.True(t, ok, "params must decode as tools.BashPermissionsParams, got %T", got)
39+ require.Equal(t, "list files", v.Description)
40+ require.Equal(t, "ls -la", v.Command)
41+ require.Equal(t, "/tmp", v.WorkingDir)
42+ },
43+ },
44+ {
45+ name: "edit",
46+ toolName: tools.EditToolName,
47+ params: tools.EditPermissionsParams{
48+ FilePath: "/tmp/x.go",
49+ OldContent: "old",
50+ NewContent: "new",
51+ },
52+ assert: func(t *testing.T, got any) {
53+ v, ok := got.(tools.EditPermissionsParams)
54+ require.True(t, ok, "params must decode as tools.EditPermissionsParams, got %T", got)
55+ require.Equal(t, "/tmp/x.go", v.FilePath)
56+ require.Equal(t, "old", v.OldContent)
57+ require.Equal(t, "new", v.NewContent)
58+ },
59+ },
60+ {
61+ name: "write",
62+ toolName: tools.WriteToolName,
63+ params: tools.WritePermissionsParams{
64+ FilePath: "/tmp/x.go",
65+ NewContent: "new",
66+ },
67+ assert: func(t *testing.T, got any) {
68+ v, ok := got.(tools.WritePermissionsParams)
69+ require.True(t, ok, "params must decode as tools.WritePermissionsParams, got %T", got)
70+ require.Equal(t, "/tmp/x.go", v.FilePath)
71+ require.Equal(t, "new", v.NewContent)
72+ },
73+ },
74+ {
75+ name: "multiedit",
76+ toolName: tools.MultiEditToolName,
77+ params: tools.MultiEditPermissionsParams{
78+ FilePath: "/tmp/x.go",
79+ OldContent: "old",
80+ NewContent: "new",
81+ },
82+ assert: func(t *testing.T, got any) {
83+ v, ok := got.(tools.MultiEditPermissionsParams)
84+ require.True(t, ok, "params must decode as tools.MultiEditPermissionsParams, got %T", got)
85+ require.Equal(t, "/tmp/x.go", v.FilePath)
86+ },
87+ },
88+ {
89+ name: "ls",
90+ toolName: tools.LSToolName,
91+ params: tools.LSPermissionsParams{
92+ Path: "/tmp",
93+ Ignore: []string{".git"},
94+ Depth: 2,
95+ },
96+ assert: func(t *testing.T, got any) {
97+ v, ok := got.(tools.LSPermissionsParams)
98+ require.True(t, ok, "params must decode as tools.LSPermissionsParams, got %T", got)
99+ require.Equal(t, "/tmp", v.Path)
100+ require.Equal(t, []string{".git"}, v.Ignore)
101+ require.Equal(t, 2, v.Depth)
102+ },
103+ },
104+ {
105+ name: "view",
106+ toolName: tools.ViewToolName,
107+ params: tools.ViewPermissionsParams{
108+ FilePath: "/tmp/x.go",
109+ Offset: 10,
110+ Limit: 100,
111+ },
112+ assert: func(t *testing.T, got any) {
113+ v, ok := got.(tools.ViewPermissionsParams)
114+ require.True(t, ok, "params must decode as tools.ViewPermissionsParams, got %T", got)
115+ require.Equal(t, "/tmp/x.go", v.FilePath)
116+ },
117+ },
118+ {
119+ name: "fetch",
120+ toolName: tools.FetchToolName,
121+ params: tools.FetchPermissionsParams{
122+ URL: "https://example.com",
123+ Format: "text",
124+ },
125+ assert: func(t *testing.T, got any) {
126+ v, ok := got.(tools.FetchPermissionsParams)
127+ require.True(t, ok, "params must decode as tools.FetchPermissionsParams, got %T", got)
128+ require.Equal(t, "https://example.com", v.URL)
129+ },
130+ },
131+ {
132+ name: "download",
133+ toolName: tools.DownloadToolName,
134+ params: tools.DownloadPermissionsParams{
135+ URL: "https://example.com/x.zip",
136+ FilePath: "/tmp/x.zip",
137+ Timeout: 30,
138+ },
139+ assert: func(t *testing.T, got any) {
140+ v, ok := got.(tools.DownloadPermissionsParams)
141+ require.True(t, ok, "params must decode as tools.DownloadPermissionsParams, got %T", got)
142+ require.Equal(t, "https://example.com/x.zip", v.URL)
143+ require.Equal(t, "/tmp/x.zip", v.FilePath)
144+ },
145+ },
146+ {
147+ name: "agentic_fetch",
148+ toolName: tools.AgenticFetchToolName,
149+ params: tools.AgenticFetchPermissionsParams{
150+ URL: "https://example.com",
151+ Prompt: "summarize this page",
152+ },
153+ assert: func(t *testing.T, got any) {
154+ v, ok := got.(tools.AgenticFetchPermissionsParams)
155+ require.True(t, ok, "params must decode as tools.AgenticFetchPermissionsParams, got %T", got)
156+ require.Equal(t, "https://example.com", v.URL)
157+ require.Equal(t, "summarize this page", v.Prompt)
158+ },
159+ },
160+ }
161+
162+ for _, tc := range tests {
163+ t.Run(tc.name, func(t *testing.T) {
164+ t.Parallel()
165+
166+ // Build a server-side request with the tool's concrete
167+ // params type, marshal to JSON (the wire path), then
168+ // decode back through proto.PermissionRequest.
169+ outbound := proto.PermissionRequest{
170+ ID: "perm-1",
171+ SessionID: "sess-1",
172+ ToolCallID: "call-1",
173+ ToolName: tc.toolName,
174+ Path: "/tmp",
175+ Params: tc.params,
176+ }
177+ data, err := json.Marshal(outbound)
178+ require.NoError(t, err)
179+
180+ var inbound proto.PermissionRequest
181+ require.NoError(t, json.Unmarshal(data, &inbound))
182+
183+ tc.assert(t, inbound.Params)
184+ })
185+ }
186+}
internal/proto/tools.go+22−38
@@ -1,5 +1,12 @@
1package proto
1package proto
2
2
3+// The wire schema for per-tool permission parameters is owned by the
4+// tool itself, not duplicated here. We alias the canonical types so
5+// there is exactly one source of truth and so values survive a
6+// round-trip across the client/server boundary as the same Go type
7+// the UI asserts on.
8+import "github.com/charmbracelet/crush/internal/agent/tools"
9+
3// ToolResponseType represents the type of tool response.
10// ToolResponseType represents the type of tool response.
4type ToolResponseType string
11type ToolResponseType string
5
12
@@ -25,10 +32,7 @@ type BashParams struct {
25}
32}
26
33
27// BashPermissionsParams represents the permission parameters for the bash tool.
34// BashPermissionsParams represents the permission parameters for the bash tool.
28−type BashPermissionsParams struct {
35+type BashPermissionsParams = tools.BashPermissionsParams
29− Command string `json:"command"`
30− Timeout int `json:"timeout"`
31−}
32
36
33// BashResponseMetadata represents the metadata for a bash tool response.
37// BashResponseMetadata represents the metadata for a bash tool response.
34type BashResponseMetadata struct {
38type BashResponseMetadata struct {
@@ -53,11 +57,7 @@ type DownloadParams struct {
53}
57}
54
58
55// DownloadPermissionsParams represents the permission parameters for the download tool.
59// DownloadPermissionsParams represents the permission parameters for the download tool.
56−type DownloadPermissionsParams struct {
60+type DownloadPermissionsParams = tools.DownloadPermissionsParams
57− URL string `json:"url"`
58− FilePath string `json:"file_path"`
59− Timeout int `json:"timeout,omitempty"`
60−}
61
61
62const EditToolName = "edit"
62const EditToolName = "edit"
63
63
@@ -70,11 +70,7 @@ type EditParams struct {
70}
70}
71
71
72// EditPermissionsParams represents the permission parameters for the edit tool.
72// EditPermissionsParams represents the permission parameters for the edit tool.
73−type EditPermissionsParams struct {
73+type EditPermissionsParams = tools.EditPermissionsParams
74− FilePath string `json:"file_path"`
75− OldContent string `json:"old_content,omitempty"`
76− NewContent string `json:"new_content,omitempty"`
77−}
78
74
79// EditResponseMetadata represents the metadata for an edit tool response.
75// EditResponseMetadata represents the metadata for an edit tool response.
80type EditResponseMetadata struct {
76type EditResponseMetadata struct {
@@ -94,11 +90,14 @@ type FetchParams struct {
94}
90}
95
91
96// FetchPermissionsParams represents the permission parameters for the fetch tool.
92// FetchPermissionsParams represents the permission parameters for the fetch tool.
97−type FetchPermissionsParams struct {
93+type FetchPermissionsParams = tools.FetchPermissionsParams
98− URL string `json:"url"`
94+
99− Format string `json:"format"`
95+// AgenticFetchToolName is the name of the agentic_fetch tool.
100− Timeout int `json:"timeout,omitempty"`
96+const AgenticFetchToolName = tools.AgenticFetchToolName
101−}
97+
98+// AgenticFetchPermissionsParams represents the permission parameters for the
99+// agentic_fetch tool.
100+type AgenticFetchPermissionsParams = tools.AgenticFetchPermissionsParams
102
101
103const GlobToolName = "glob"
102const GlobToolName = "glob"
104
103
@@ -139,10 +138,7 @@ type LSParams struct {
139}
138}
140
139
141// LSPermissionsParams represents the permission parameters for the ls tool.
140// LSPermissionsParams represents the permission parameters for the ls tool.
142−type LSPermissionsParams struct {
141+type LSPermissionsParams = tools.LSPermissionsParams
143− Path string `json:"path"`
144− Ignore []string `json:"ignore"`
145−}
146
142
147// TreeNode represents a node in a directory tree.
143// TreeNode represents a node in a directory tree.
148type TreeNode struct {
144type TreeNode struct {
@@ -174,11 +170,7 @@ type MultiEditParams struct {
174}
170}
175
171
176// MultiEditPermissionsParams represents the permission parameters for the multi-edit tool.
172// MultiEditPermissionsParams represents the permission parameters for the multi-edit tool.
177−type MultiEditPermissionsParams struct {
173+type MultiEditPermissionsParams = tools.MultiEditPermissionsParams
178− FilePath string `json:"file_path"`
179− OldContent string `json:"old_content,omitempty"`
180− NewContent string `json:"new_content,omitempty"`
181−}
182
174
183// MultiEditResponseMetadata represents the metadata for a multi-edit tool response.
175// MultiEditResponseMetadata represents the metadata for a multi-edit tool response.
184type MultiEditResponseMetadata struct {
176type MultiEditResponseMetadata struct {
@@ -215,11 +207,7 @@ type ViewParams struct {
215}
207}
216
208
217// ViewPermissionsParams represents the permission parameters for the view tool.
209// ViewPermissionsParams represents the permission parameters for the view tool.
218−type ViewPermissionsParams struct {
210+type ViewPermissionsParams = tools.ViewPermissionsParams
219− FilePath string `json:"file_path"`
220− Offset int `json:"offset"`
221− Limit int `json:"limit"`
222−}
223
211
224// ViewResponseMetadata represents the metadata for a view tool response.
212// ViewResponseMetadata represents the metadata for a view tool response.
225type ViewResponseMetadata struct {
213type ViewResponseMetadata struct {
@@ -236,11 +224,7 @@ type WriteParams struct {
236}
224}
237
225
238// WritePermissionsParams represents the permission parameters for the write tool.
226// WritePermissionsParams represents the permission parameters for the write tool.
239−type WritePermissionsParams struct {
227+type WritePermissionsParams = tools.WritePermissionsParams
240− FilePath string `json:"file_path"`
241− OldContent string `json:"old_content,omitempty"`
242− NewContent string `json:"new_content,omitempty"`
243−}
244
228
245// WriteResponseMetadata represents the metadata for a write tool response.
229// WriteResponseMetadata represents the metadata for a write tool response.
246type WriteResponseMetadata struct {
230type WriteResponseMetadata struct {
internal/server/events.go+3
@@ -186,6 +186,9 @@ func messageToProto(m message.Message) proto.Message {
186 ToolCallID: v.ToolCallID,
186 ToolCallID: v.ToolCallID,
187 Name: v.Name,
187 Name: v.Name,
188 Content: v.Content,
188 Content: v.Content,
189+ Data: v.Data,
190+ MIMEType: v.MIMEType,
191+ Metadata: v.Metadata,
189 IsError: v.IsError,
192 IsError: v.IsError,
190 })
193 })
191 case message.Finish:
194 case message.Finish:
internal/workspace/client_workspace.go+3
@@ -722,6 +722,9 @@ func protoToMessage(m proto.Message) message.Message {
722 ToolCallID: v.ToolCallID,
722 ToolCallID: v.ToolCallID,
723 Name: v.Name,
723 Name: v.Name,
724 Content: v.Content,
724 Content: v.Content,
725+ Data: v.Data,
726+ MIMEType: v.MIMEType,
727+ Metadata: v.Metadata,
725 IsError: v.IsError,
728 IsError: v.IsError,
726 })
729 })
727 case proto.Finish:
730 case proto.Finish: