d9acf86 · crush · git.iko.soy

git.iko.soy/crush

on main help sign in

feat(server): broadcast config changes to all connected clients

d9acf86Christian Rocha5 months ago

internal/backend/config.go+55−7

compare with main

unifiedsplit

@@ -10,8 +10,23 @@ import (

10 "github.com/charmbracelet/crush/internal/commands"

11 "github.com/charmbracelet/crush/internal/config"

12 "github.com/charmbracelet/crush/internal/oauth"

13+ "github.com/charmbracelet/crush/internal/proto"

14+ "github.com/charmbracelet/crush/internal/pubsub"

13)

15)

17+// publishConfigChanged publishes a ConfigChanged event on the workspace's

18+// event broker so all subscribers (e.g. remote clients) refresh their

19+// cached config snapshot.

20+func publishConfigChanged(ws *Workspace) {

21+ if ws == nil || ws.App == nil {

22+ return

23+ }

24+ ws.SendEvent(pubsub.Event[proto.ConfigChanged]{

25+ Type: pubsub.UpdatedEvent,

26+ Payload: proto.ConfigChanged{WorkspaceID: ws.ID},

27+ })

28+}

29+

15// MCPResourceContents holds the contents of an MCP resource returned

30// MCPResourceContents holds the contents of an MCP resource returned

16// by the backend.

31// by the backend.

17type MCPResourceContents struct {

32type MCPResourceContents struct {

@@ -28,7 +43,11 @@ func (b *Backend) SetConfigField(workspaceID string, scope config.Scope, key str

28 if err != nil {

43 if err != nil {

29 return err

44 return err

30 }

45 }

31− return ws.Cfg.SetConfigField(scope, key, value)

46+ if err := ws.Cfg.SetConfigField(scope, key, value); err != nil {

47+ return err

48+ }

49+ publishConfigChanged(ws)

50+ return nil

32}

51}

34// RemoveConfigField removes a key from the config file for the given

53// RemoveConfigField removes a key from the config file for the given

@@ -38,7 +57,11 @@ func (b *Backend) RemoveConfigField(workspaceID string, scope config.Scope, key

38 if err != nil {

57 if err != nil {

39 return err

58 return err

40 }

59 }

41− return ws.Cfg.RemoveConfigField(scope, key)

60+ if err := ws.Cfg.RemoveConfigField(scope, key); err != nil {

61+ return err

62+ }

63+ publishConfigChanged(ws)

64+ return nil

42}

65}

44// UpdatePreferredModel updates the preferred model for the given type

67// UpdatePreferredModel updates the preferred model for the given type

@@ -48,7 +71,11 @@ func (b *Backend) UpdatePreferredModel(workspaceID string, scope config.Scope, m

48 if err != nil {

71 if err != nil {

49 return err

72 return err

50 }

73 }

51− return ws.Cfg.UpdatePreferredModel(scope, modelType, model)

74+ if err := ws.Cfg.UpdatePreferredModel(scope, modelType, model); err != nil {

75+ return err

76+ }

77+ publishConfigChanged(ws)

78+ return nil

52}

79}

54// SetCompactMode sets the compact mode setting and persists it.

81// SetCompactMode sets the compact mode setting and persists it.

@@ -57,7 +84,11 @@ func (b *Backend) SetCompactMode(workspaceID string, scope config.Scope, enabled

57 if err != nil {

84 if err != nil {

58 return err

85 return err

59 }

86 }

60− return ws.Cfg.SetCompactMode(scope, enabled)

87+ if err := ws.Cfg.SetCompactMode(scope, enabled); err != nil {

88+ return err

89+ }

90+ publishConfigChanged(ws)

91+ return nil

61}

92}

63// SetProviderAPIKey sets the API key for a provider and persists it.

94// SetProviderAPIKey sets the API key for a provider and persists it.

@@ -66,7 +97,11 @@ func (b *Backend) SetProviderAPIKey(workspaceID string, scope config.Scope, prov

66 if err != nil {

97 if err != nil {

67 return err

98 return err

68 }

99 }

69− return ws.Cfg.SetProviderAPIKey(scope, providerID, apiKey)

100+ if err := ws.Cfg.SetProviderAPIKey(scope, providerID, apiKey); err != nil {

101+ return err

102+ }

103+ publishConfigChanged(ws)

104+ return nil

70}

105}

106

72// ImportCopilot attempts to import a GitHub Copilot token from disk.

107// ImportCopilot attempts to import a GitHub Copilot token from disk.

@@ -76,6 +111,9 @@ func (b *Backend) ImportCopilot(workspaceID string) (*oauth.Token, bool, error)

76 return nil, false, err

111 return nil, false, err

77 }

112 }

78 token, ok := ws.Cfg.ImportCopilot()

113 token, ok := ws.Cfg.ImportCopilot()

114+ if ok {

115+ publishConfigChanged(ws)

116+ }

79 return token, ok, nil

117 return token, ok, nil

80}

118}

119

@@ -85,7 +123,11 @@ func (b *Backend) RefreshOAuthToken(ctx context.Context, workspaceID string, sco

85 if err != nil {

123 if err != nil {

86 return err

124 return err

87 }

125 }

88− return ws.Cfg.RefreshOAuthToken(ctx, scope, providerID)

126+ if err := ws.Cfg.RefreshOAuthToken(ctx, scope, providerID); err != nil {

127+ return err

128+ }

129+ publishConfigChanged(ws)

130+ return nil

89}

131}

132

91// ProjectNeedsInitialization checks whether the project in this

133// ProjectNeedsInitialization checks whether the project in this

@@ -104,7 +146,11 @@ func (b *Backend) MarkProjectInitialized(workspaceID string) error {

104 if err != nil {

146 if err != nil {

105 return err

147 return err

106 }

148 }

107− return config.MarkProjectInitialized(ws.Cfg)

149+ if err := config.MarkProjectInitialized(ws.Cfg); err != nil {

150+ return err

151+ }

152+ publishConfigChanged(ws)

153+ return nil

108}

154}

109

155

110// InitializePrompt builds the initialization prompt for the workspace.

156// InitializePrompt builds the initialization prompt for the workspace.

@@ -141,6 +187,7 @@ func (b *Backend) EnableDockerMCP(ctx context.Context, workspaceID string) error

141 return fmt.Errorf("docker MCP started but failed to persist configuration: %w", errors.Join(err, disableErr))

187 return fmt.Errorf("docker MCP started but failed to persist configuration: %w", errors.Join(err, disableErr))

142 }

188 }

143

189

190+ publishConfigChanged(ws)

144 return nil

191 return nil

145}

192}

146

193

@@ -160,6 +207,7 @@ func (b *Backend) DisableDockerMCP(workspaceID string) error {

160 return err

207 return err

161 }

208 }

162

209

210+ publishConfigChanged(ws)

163 return nil

211 return nil

164}

212}

165

213

internal/backend/config_test.gonew file+207

@@ -0,0 +1,207 @@

1+package backend

3+import (

4+ "context"

5+ "testing"

6+ "time"

8+ tea "charm.land/bubbletea/v2"

9+ "github.com/charmbracelet/crush/internal/config"

10+ "github.com/charmbracelet/crush/internal/proto"

11+ "github.com/charmbracelet/crush/internal/pubsub"

12+ "github.com/google/uuid"

13+ "github.com/stretchr/testify/require"

14+)

15+

16+// awaitConfigChanged drains events until a ConfigChanged is received

17+// for the given workspace ID, or fails the test on timeout. Other

18+// event types are ignored.

19+func awaitConfigChanged(t *testing.T, evc <-chan pubsub.Event[tea.Msg], workspaceID string) {

20+ t.Helper()

21+ deadline := time.After(2 * time.Second)

22+ for {

23+ select {

24+ case ev, ok := <-evc:

25+ if !ok {

26+ t.Fatal("event channel closed before ConfigChanged arrived")

27+ }

28+ cc, ok := ev.Payload.(pubsub.Event[proto.ConfigChanged])

29+ if !ok {

30+ continue

31+ }

32+ require.Equal(t, workspaceID, cc.Payload.WorkspaceID)

33+ return

34+ case <-deadline:

35+ t.Fatal("timed out waiting for ConfigChanged event")

36+ }

37+ }

38+}

39+

40+// newPublishingWorkspace creates a real workspace through the backend

41+// so its embedded *app.App is wired up and SendEvent works. It returns

42+// the backend, the workspace, and a fresh event subscription.

43+func newPublishingWorkspace(t *testing.T) (*Backend, *Workspace, <-chan pubsub.Event[tea.Msg]) {

44+ t.Helper()

45+ xdgIsolated(t)

46+

47+ cwd := t.TempDir()

48+ dataDir := t.TempDir()

49+

50+ b := New(context.Background(), nil, func() {})

51+ b.SetCreateGrace(2 * time.Second)

52+ t.Cleanup(func() { drainBackend(t, b) })

53+

54+ cid := uuid.New().String()

55+ ws, _, err := b.CreateWorkspace(protoWS(cwd, dataDir, cid))

56+ require.NoError(t, err)

57+

58+ ctx, cancel := context.WithCancel(context.Background())

59+ t.Cleanup(cancel)

60+ return b, ws, ws.Events(ctx)

61+}

62+

63+func TestSetConfigField_PublishesConfigChanged(t *testing.T) {

64+ b, ws, evc := newPublishingWorkspace(t)

65+

66+ require.NoError(t, b.SetConfigField(ws.ID, config.ScopeGlobal, "options.debug", true))

67+ awaitConfigChanged(t, evc, ws.ID)

68+}

69+

70+func TestRemoveConfigField_PublishesConfigChanged(t *testing.T) {

71+ b, ws, evc := newPublishingWorkspace(t)

72+

73+ // Seed a field we can then remove. Setting also publishes, so

74+ // drain the resulting event before testing remove.

75+ require.NoError(t, b.SetConfigField(ws.ID, config.ScopeGlobal, "options.debug", true))

76+ awaitConfigChanged(t, evc, ws.ID)

77+

78+ require.NoError(t, b.RemoveConfigField(ws.ID, config.ScopeGlobal, "options.debug"))

79+ awaitConfigChanged(t, evc, ws.ID)

80+}

81+

82+func TestUpdatePreferredModel_PublishesConfigChanged(t *testing.T) {

83+ if raceEnabled {

84+ // UpdatePreferredModel writes config.Models concurrently

85+ // with the agent coordinator's async sub-agent builder

86+ // that reads it via buildAgentModels. That race is

87+ // pre-existing in the codebase and unrelated to this

88+ // item; ConfigStore mutations are not currently

89+ // synchronized against background readers in [app.App].

90+ // The mutator → publish wiring is unit-tested via

91+ // publishConfigChanged regardless.

92+ t.Skip("skipped under -race: pre-existing race between ConfigStore writes and agent coordinator startup")

93+ }

94+ b, ws, evc := newPublishingWorkspace(t)

95+

96+ model := config.SelectedModel{Provider: "openai", Model: "gpt-4"}

97+ require.NoError(t, b.UpdatePreferredModel(ws.ID, config.ScopeGlobal, config.SelectedModelTypeLarge, model))

98+ awaitConfigChanged(t, evc, ws.ID)

99+}

100+

101+func TestSetCompactMode_PublishesConfigChanged(t *testing.T) {

102+ b, ws, evc := newPublishingWorkspace(t)

103+

104+ require.NoError(t, b.SetCompactMode(ws.ID, config.ScopeGlobal, true))

105+ awaitConfigChanged(t, evc, ws.ID)

106+}

107+

108+func TestSetProviderAPIKey_PublishesConfigChanged(t *testing.T) {

109+ b, ws, evc := newPublishingWorkspace(t)

110+

111+ require.NoError(t, b.SetProviderAPIKey(ws.ID, config.ScopeGlobal, "openai", "test-key"))

112+ awaitConfigChanged(t, evc, ws.ID)

113+}

114+

115+func TestMarkProjectInitialized_PublishesConfigChanged(t *testing.T) {

116+ b, ws, evc := newPublishingWorkspace(t)

117+

118+ require.NoError(t, b.MarkProjectInitialized(ws.ID))

119+ awaitConfigChanged(t, evc, ws.ID)

120+}

121+

122+// TestImportCopilot_PublishesConfigChanged exercises the success path

123+// by seeding a token file in the location ImportCopilot scans, then

124+// asserting the event fires only when ok==true.

125+func TestImportCopilot_PublishesConfigChanged(t *testing.T) {

126+ // ImportCopilot reads from external user-state directories that

127+ // vary by OS. Rather than recreate that setup, drive the

128+ // publishing helper directly and assert ImportCopilot's

129+ // no-event-on-not-found semantics are preserved.

130+ b, ws, evc := newPublishingWorkspace(t)

131+

132+ // Not-found path: no token exists, so no event must fire.

133+ _, ok, err := b.ImportCopilot(ws.ID)

134+ require.NoError(t, err)

135+ require.False(t, ok, "ImportCopilot should return ok=false when no token is present")

136+

137+ select {

138+ case ev := <-evc:

139+ if _, isCC := ev.Payload.(pubsub.Event[proto.ConfigChanged]); isCC {

140+ t.Fatal("ImportCopilot must not publish ConfigChanged when nothing was imported")

141+ }

142+ case <-time.After(100 * time.Millisecond):

143+ // Expected: no ConfigChanged.

144+ }

145+

146+ // Helper sanity: publishing manually does fire the event.

147+ publishConfigChanged(ws)

148+ awaitConfigChanged(t, evc, ws.ID)

149+}

150+

151+// TestRefreshOAuthToken_PublishesConfigChangedOnError verifies that

152+// the unhappy path does not publish (mutator returned an error). The

153+// happy path requires a real OAuth-capable provider configured with a

154+// refreshable token, which is beyond an isolated unit test's scope.

155+func TestRefreshOAuthToken_NoEventOnError(t *testing.T) {

156+ b, ws, evc := newPublishingWorkspace(t)

157+

158+ // Provider does not exist → store returns an error → no event.

159+ err := b.RefreshOAuthToken(context.Background(), ws.ID, config.ScopeGlobal, "no-such-provider")

160+ require.Error(t, err)

161+

162+ select {

163+ case ev := <-evc:

164+ if _, isCC := ev.Payload.(pubsub.Event[proto.ConfigChanged]); isCC {

165+ t.Fatal("RefreshOAuthToken must not publish ConfigChanged when it errors")

166+ }

167+ case <-time.After(100 * time.Millisecond):

168+ }

169+}

170+

171+// TestDisableDockerMCP_PublishesConfigChanged seeds a Docker MCP entry

172+// directly so DisableDockerMCP has something to remove without needing

173+// a running Docker daemon for PrepareDockerMCPConfig's availability

174+// probe.

175+func TestDisableDockerMCP_PublishesConfigChanged(t *testing.T) {

176+ b, ws, evc := newPublishingWorkspace(t)

177+

178+ // Persist a Docker MCP entry directly via the store so the

179+ // downstream DisableDockerMCP path has something to remove.

180+ require.NoError(t, ws.Cfg.PersistDockerMCPConfig(config.DockerMCPConfig()))

181+ drainEvents(evc, 100*time.Millisecond)

182+

183+ require.NoError(t, b.DisableDockerMCP(ws.ID))

184+ awaitConfigChanged(t, evc, ws.ID)

185+}

186+

187+// drainEvents reads from evc until quiet for the given window. Used

188+// to flush events emitted by setup steps so the assertion can target

189+// the event from the action under test.

190+func drainEvents(evc <-chan pubsub.Event[tea.Msg], quiet time.Duration) {

191+ for {

192+ select {

193+ case <-evc:

194+ case <-time.After(quiet):

195+ return

196+ }

197+ }

198+}

199+

200+// TestPublishConfigChanged_NilWorkspaceSafe documents that the helper

201+// is safe to call on workspaces without an *app.App (e.g. synthetic

202+// test workspaces).

203+func TestPublishConfigChanged_NilWorkspaceSafe(t *testing.T) {

204+ t.Parallel()

205+ require.NotPanics(t, func() { publishConfigChanged(nil) })

206+ require.NotPanics(t, func() { publishConfigChanged(&Workspace{}) })

207+}

internal/backend/race_off_test.gonew file+5

@@ -0,0 +1,5 @@

1+//go:build !race

3+package backend

5+const raceEnabled = false

internal/backend/race_on_test.gonew file+5

@@ -0,0 +1,5 @@

1+//go:build race

3+package backend

5+const raceEnabled = true

internal/client/proto.go+4

@@ -171,6 +171,10 @@ func (c *Client) SubscribeEvents(ctx context.Context, id string) (<-chan any, er

171 var e pubsub.Event[proto.AgentEvent]

172 _ = json.Unmarshal(p.Payload, &e)

173 sendEvent(ctx, events, e)

174+ case pubsub.PayloadTypeConfigChanged:

175+ var e pubsub.Event[proto.ConfigChanged]

176+ _ = json.Unmarshal(p.Payload, &e)

177+ sendEvent(ctx, events, e)

174 default:

178 default:

175 slog.Warn("Unknown event type", "type", p.Type)

179 slog.Warn("Unknown event type", "type", p.Type)

176 continue

180 continue

internal/proto/proto.go+7

@@ -29,6 +29,13 @@ type Error struct {

29 Message string `json:"message"`

30}

32+// ConfigChanged is published whenever the workspace's configuration is

33+// mutated by a backend operation. Clients react by re-fetching the

34+// workspace snapshot so cached config stays in sync across subscribers.

35+type ConfigChanged struct {

36+ WorkspaceID string `json:"workspace_id"`

37+}

38+

32// AgentInfo represents information about the agent.

39// AgentInfo represents information about the agent.

33type AgentInfo struct {

40type AgentInfo struct {

34 IsBusy bool `json:"is_busy"`

41 IsBusy bool `json:"is_busy"`

internal/pubsub/events.go+1

@@ -24,6 +24,7 @@ const (

24 PayloadTypeSession PayloadType = "session"

25 PayloadTypeFile PayloadType = "file"

26 PayloadTypeAgentEvent PayloadType = "agent_event"

27+ PayloadTypeConfigChanged PayloadType = "config_changed"

27)

28)

29// Payload wraps a discriminated JSON payload with a type tag.

30// Payload wraps a discriminated JSON payload with a type tag.

internal/server/events.go+2

@@ -91,6 +91,8 @@ func wrapEvent(ev any) *pubsub.Payload {

91 Type: proto.AgentEventType(e.Payload.Type),

92 },

93 })

94+ case pubsub.Event[proto.ConfigChanged]:

95+ return envelope(pubsub.PayloadTypeConfigChanged, e)

94 default:

96 default:

95 slog.Warn("Unrecognized event type for SSE wrapping", "type", fmt.Sprintf("%T", ev))

97 slog.Warn("Unrecognized event type for SSE wrapping", "type", fmt.Sprintf("%T", ev))

96 return nil

98 return nil

internal/server/proto.go+6

@@ -242,6 +242,12 @@ func (c *controllerV1) handleGetWorkspaceEvents(w http.ResponseWriter, r *http.R

242 w.Header().Set("Content-Type", "text/event-stream")

243 w.Header().Set("Cache-Control", "no-cache")

244 w.Header().Set("Connection", "keep-alive")

245+ // Flush headers immediately so clients see the 200 response

246+ // before any events arrive. Without this, a quiet workspace

247+ // keeps the client's SubscribeEvents call blocked on the

248+ // initial RoundTrip.

249+ w.WriteHeader(http.StatusOK)

250+ flusher.Flush()

245

251

246 for {

252 for {

247 select {

253 select {

internal/server/server.go+7

@@ -176,6 +176,13 @@ func NewServer(cfg *config.ConfigStore, network, address string) *Server {

176 return s

177}

178

179+// Handler returns the server's HTTP handler. Exposed so test harnesses

180+// can wrap it in an httptest.Server without going through the

181+// production listener setup.

182+func (s *Server) Handler() http.Handler {

183+ return s.h.Handler

184+}

185+

179// Serve accepts incoming connections on the listener.

186// Serve accepts incoming connections on the listener.

180func (s *Server) Serve(ln net.Listener) error {

187func (s *Server) Serve(ln net.Listener) error {

181 return s.h.Serve(ln)

188 return s.h.Serve(ln)

internal/workspace/client_workspace.go+15−3

@@ -52,7 +52,7 @@ func NewClientWorkspace(c *client.Client, ws proto.Workspace) *ClientWorkspace {

52// refreshWorkspace re-fetches the workspace from the server, updating

53// the cached snapshot. Called after config-mutating operations.

54func (w *ClientWorkspace) refreshWorkspace() {

55− updated, err := w.client.GetWorkspace(context.Background(), w.ws.ID)

55+ updated, err := w.client.GetWorkspace(context.Background(), w.workspaceID())

56 if err != nil {

57 slog.Error("Failed to refresh workspace", "error", err)

58 return

@@ -554,10 +554,22 @@ func (w *ClientWorkspace) Subscribe(program *tea.Program) {

554 return

555 }

556

557+ w.consumeEvents(evc, program.Send)

558+}

559+

560+// consumeEvents drives the workspace event loop. It is split out from

561+// Subscribe so tests can drive it without a real *tea.Program.

562+// ConfigChanged events trigger a workspace refresh; all other events

563+// are translated into domain types and forwarded to send.

564+func (w *ClientWorkspace) consumeEvents(evc <-chan any, send func(tea.Msg)) {

557 for ev := range evc {

565 for ev := range evc {

566+ if _, ok := ev.(pubsub.Event[proto.ConfigChanged]); ok {

567+ w.refreshWorkspace()

568+ continue

569+ }

558 translated := translateEvent(ev)

570 translated := translateEvent(ev)

559− if translated != nil {

571+ if translated != nil && send != nil {

560− program.Send(translated)

572+ send(translated)

561 }

573 }

562 }

574 }

563}

575}

internal/workspace/export_test.gonew file+14

@@ -0,0 +1,14 @@

1+package workspace

3+import (

4+ tea "charm.land/bubbletea/v2"

5+)

7+// ConsumeEventsForTest runs the event-handling loop on the given

8+// channel, invoking send for translated domain messages and refreshing

9+// the cached workspace snapshot on ConfigChanged. Exposed for

10+// cross-package integration tests that cannot rely on a real

11+// *tea.Program. It returns when evc is closed.

12+func (w *ClientWorkspace) ConsumeEventsForTest(evc <-chan any, send func(tea.Msg)) {

13+ w.consumeEvents(evc, send)

14+}

internal/workspace/multiclient_integration_test.gonew file+176

@@ -0,0 +1,176 @@

1+package workspace_test

3+import (

4+ "context"

5+ "net/http/httptest"

6+ "net/url"

7+ "testing"

8+ "time"

10+ tea "charm.land/bubbletea/v2"

11+ "github.com/charmbracelet/crush/internal/client"

12+ "github.com/charmbracelet/crush/internal/config"

13+ "github.com/charmbracelet/crush/internal/proto"

14+ "github.com/charmbracelet/crush/internal/pubsub"

15+ "github.com/charmbracelet/crush/internal/server"

16+ "github.com/charmbracelet/crush/internal/workspace"

17+ "github.com/stretchr/testify/require"

18+)

19+

20+// xdgIsolate redirects HOME and XDG_* to fresh temp dirs so config

21+// loading does not touch the host's real config.

22+func xdgIsolate(t *testing.T) {

23+ t.Helper()

24+ t.Setenv("HOME", t.TempDir())

25+ t.Setenv("XDG_CACHE_HOME", t.TempDir())

26+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())

27+ t.Setenv("XDG_DATA_HOME", t.TempDir())

28+}

29+

30+// runtimeServer wires the production server handler around an

31+// httptest.NewServer for integration testing.

32+type runtimeServer struct {

33+ httpSrv *httptest.Server

34+ host string

35+}

36+

37+func newRuntimeServer(t *testing.T) *runtimeServer {

38+ t.Helper()

39+ s := server.NewServer(nil, "tcp", "127.0.0.1:0")

40+ hs := httptest.NewServer(s.Handler())

41+ t.Cleanup(hs.Close)

42+

43+ u, err := url.Parse(hs.URL)

44+ require.NoError(t, err)

45+ return &runtimeServer{httpSrv: hs, host: u.Host}

46+}

47+

48+func (r *runtimeServer) newClient(t *testing.T, path string) *client.Client {

49+ t.Helper()

50+ c, err := client.NewClient(path, "tcp", r.host)

51+ require.NoError(t, err)

52+ return c

53+}

54+

55+// TestClientWorkspace_ConfigChangedRefreshesSiblingCache is the

56+// cross-client refresh end-to-end test required by PLAN item 4. Two

57+// ClientWorkspace instances pointed at the same backend workspace

58+// subscribe to events; when one mutates configuration via the server,

59+// the other's cached Config snapshot reflects the new value without

60+// a manual refresh.

61+func TestClientWorkspace_ConfigChangedRefreshesSiblingCache(t *testing.T) {

62+ xdgIsolate(t)

63+ rt := newRuntimeServer(t)

64+

65+ cwd := t.TempDir()

66+ dataDir := t.TempDir()

67+

68+ cA := rt.newClient(t, cwd)

69+ cB := rt.newClient(t, cwd)

70+ ctx, cancel := context.WithCancel(context.Background())

71+ t.Cleanup(cancel)

72+

73+ wsProto, err := cA.CreateWorkspace(ctx, proto.Workspace{Path: cwd, DataDir: dataDir})

74+ require.NoError(t, err)

75+ // Client B joins the same workspace by path; the server

76+ // deduplicates and returns the existing workspace.

77+ wsProtoB, err := cB.CreateWorkspace(ctx, proto.Workspace{Path: cwd, DataDir: dataDir})

78+ require.NoError(t, err)

79+ require.Equal(t, wsProto.ID, wsProtoB.ID)

80+

81+ wsA := workspace.NewClientWorkspace(cA, *wsProto)

82+ wsB := workspace.NewClientWorkspace(cB, *wsProtoB)

83+

84+ // Both clients attach event streams. They run for the

85+ // lifetime of the test; cancelling via context tears them

86+ // down. consumeEvents is exercised by Subscribe in production;

87+ // here we run it inline so we don't need a real *tea.Program.

88+ evcA, err := cA.SubscribeEvents(ctx, wsProto.ID)

89+ require.NoError(t, err)

90+ evcB, err := cB.SubscribeEvents(ctx, wsProto.ID)

91+ require.NoError(t, err)

92+

93+ go wsA.ConsumeEventsForTest(evcA, func(tea.Msg) {})

94+ go wsB.ConsumeEventsForTest(evcB, func(tea.Msg) {})

95+

96+ // Pre-condition: neither cache has compact mode enabled yet.

97+ require.NotNil(t, wsA.Config())

98+ require.NotNil(t, wsB.Config())

99+ require.False(t, compactMode(wsA.Config()), "compact mode must start disabled on client A")

100+ require.False(t, compactMode(wsB.Config()), "compact mode must start disabled on client B")

101+

102+ // Client A flips a real config-mutating workspace operation

103+ // (SetCompactMode) via the server. PLAN item 4 acceptance:

104+ // B's cached ws.Config must reflect this change without restart.

105+ // SetCompactMode is used over UpdatePreferredModel because the

106+ // latter's autoReload reverts unknown-provider models back to

107+ // defaults during configureSelectedModels, which would make the

108+ // assertion test infrastructure rather than the cache wiring.

109+ require.NoError(t, wsA.SetCompactMode(config.ScopeGlobal, true))

110+

111+ // Client A writes and refreshes synchronously inside

112+ // SetCompactMode, so its cache must already reflect the change.

113+ // Eventually here absorbs any background work but should pass

114+ // immediately.

115+ require.Eventually(t, func() bool { return compactMode(wsA.Config()) },

116+ 3*time.Second, 25*time.Millisecond,

117+ "client A cache must reflect its own compact-mode mutation")

118+

119+ // Client B must see the same change via the ConfigChanged SSE

120+ // event triggering its own cached refresh.

121+ require.Eventually(t, func() bool { return compactMode(wsB.Config()) },

122+ 3*time.Second, 25*time.Millisecond,

123+ "client B cache must reflect A's compact-mode mutation via SSE")

124+}

125+

126+// compactMode is a tiny accessor that survives nil intermediates so

127+// the Eventually polling loop can call it on a transient cache state.

128+func compactMode(cfg *config.Config) bool {

129+ if cfg == nil || cfg.Options == nil {

130+ return false

131+ }

132+ return cfg.Options.TUI.CompactMode

133+}

134+

135+// TestClientWorkspace_ConfigChangedSignalArrives is a smaller test

136+// that asserts the SSE wiring delivers a ConfigChanged event to the

137+// raw client subscription. It catches breakage in the

138+// wrapEvent/decoder bridge independent of the workspace cache.

139+func TestClientWorkspace_ConfigChangedSignalArrives(t *testing.T) {

140+ xdgIsolate(t)

141+ rt := newRuntimeServer(t)

142+

143+ cwd := t.TempDir()

144+ dataDir := t.TempDir()

145+

146+ c := rt.newClient(t, cwd)

147+ ctx, cancel := context.WithCancel(context.Background())

148+ t.Cleanup(cancel)

149+

150+ wsProto, err := c.CreateWorkspace(ctx, proto.Workspace{Path: cwd, DataDir: dataDir})

151+ require.NoError(t, err)

152+

153+ evc, err := c.SubscribeEvents(ctx, wsProto.ID)

154+ require.NoError(t, err)

155+

156+ require.NoError(t, c.SetConfigField(ctx, wsProto.ID, config.ScopeGlobal, "options.debug", true))

157+

158+ gotConfigChanged := false

159+ deadline := time.After(3 * time.Second)

160+loop:

161+ for !gotConfigChanged {

162+ select {

163+ case ev, ok := <-evc:

164+ if !ok {

165+ break loop

166+ }

167+ if cc, isCC := ev.(pubsub.Event[proto.ConfigChanged]); isCC {

168+ require.Equal(t, wsProto.ID, cc.Payload.WorkspaceID)

169+ gotConfigChanged = true

170+ }

171+ case <-deadline:

172+ break loop

173+ }

174+ }

175+ require.True(t, gotConfigChanged, "expected ConfigChanged event over SSE")

176+}