@@ -1,170 +1,1241 @@
1−package backend_test
1+package backend
22
33import (
4+ "bytes"
45 "context"
5− "fmt"
6+ "errors"
7+ "log/slog"
68 "os"
79 "path/filepath"
10+ "runtime"
11+ "strings"
12+ "sync"
13+ "sync/atomic"
814 "testing"
915 "time"
1016
11− tea "charm.land/bubbletea/v2"
12− "github.com/charmbracelet/crush/internal/backend"
13− "github.com/charmbracelet/crush/internal/config"
17+ "github.com/charmbracelet/crush/internal/csync"
1418 "github.com/charmbracelet/crush/internal/proto"
15− "github.com/charmbracelet/crush/internal/pubsub"
16− "github.com/charmbracelet/crush/internal/skills"
19+ "github.com/google/uuid"
1720 "github.com/stretchr/testify/require"
1821)
1922
20−// TestBackend_WorkspaceSkillsIsolation verifies that skill discovery
21−// state and SSE events are per-workspace, not process-global. This
22−// guards the structural change in §5 of the plan: two workspaces in the
23−// same backend process must not see each other's discoveries (either in
24−// their initial snapshot or in subsequent PublishStates events).
25−//
26−// Before that change landed, the package-level latestStates cache and
27−// the package-level broker meant that:
28−// - workspace A's PublishStates would arrive on workspace B's SSE
29−// stream, and
30−// - the most recent SetLatestStates would silently overwrite the
31−// other workspace's cache.
32−//
33−// This test fails on either scenario.
34−func TestBackend_WorkspaceSkillsIsolation(t *testing.T) {
35− // Isolate all of config.Init's filesystem reads from the host. The
36− // project-local .agents/skills/<name>/SKILL.md per working dir is
37− // what we actually want each workspace to see; everything else
38− // (global skills, XDG dirs, etc.) must be empty/deterministic.
39− hostHome := t.TempDir()
40− t.Setenv("HOME", hostHome)
41− t.Setenv("XDG_CONFIG_HOME", filepath.Join(hostHome, ".config"))
42− t.Setenv("XDG_DATA_HOME", filepath.Join(hostHome, ".local", "share"))
43− t.Setenv("XDG_CACHE_HOME", filepath.Join(hostHome, ".cache"))
44− t.Setenv("CRUSH_SKILLS_DIR", t.TempDir())
45−
46− // Each workspace gets its own working directory containing a
47− // distinct project-local skill so the discovery output differs.
48− wdA := t.TempDir()
49− wdB := t.TempDir()
50− writeSkill(t, wdA, "wsa-only-skill", "Workspace A only skill.")
51− writeSkill(t, wdB, "wsb-only-skill", "Workspace B only skill.")
52−
53− srvCfg, err := config.Init(wdA, "", false)
54− require.NoError(t, err)
55− b := backend.New(t.Context(), srvCfg, nil)
56−
57− wsA, _, err := b.CreateWorkspace(proto.Workspace{
58− Path: wdA,
59− DataDir: filepath.Join(wdA, ".crush"),
60− })
61− require.NoError(t, err)
62− t.Cleanup(func() { b.DeleteWorkspace(wsA.ID) })
63−
64− wsB, _, err := b.CreateWorkspace(proto.Workspace{
65− Path: wdB,
66− DataDir: filepath.Join(wdB, ".crush"),
67− })
68− require.NoError(t, err)
69− t.Cleanup(func() { b.DeleteWorkspace(wsB.ID) })
70−
71− require.NotNil(t, wsA.Skills, "workspace A must have its own skills.Manager")
72− require.NotNil(t, wsB.Skills, "workspace B must have its own skills.Manager")
73− require.NotSame(t, wsA.Skills, wsB.Skills, "managers must be distinct instances per workspace")
74−
75− // Initial snapshots see each workspace's own filesystem skill, and
76− // neither sees the other's.
77− statesA := wsA.Skills.States()
78− statesB := wsB.Skills.States()
79− require.True(t, containsSkillName(statesA, "wsa-only-skill"),
80− "workspace A snapshot missing its own skill")
81− require.False(t, containsSkillName(statesA, "wsb-only-skill"),
82− "workspace A snapshot leaked workspace B's skill")
83− require.True(t, containsSkillName(statesB, "wsb-only-skill"),
84− "workspace B snapshot missing its own skill")
85− require.False(t, containsSkillName(statesB, "wsa-only-skill"),
86− "workspace B snapshot leaked workspace A's skill")
87−
88− // Subscribe to each workspace's SSE event stream.
89− ctxA, cancelA := context.WithCancel(t.Context())
90− t.Cleanup(cancelA)
91− chA, err := b.SubscribeEvents(ctxA, wsA.ID)
92− require.NoError(t, err)
93−
94− ctxB, cancelB := context.WithCancel(t.Context())
95− t.Cleanup(cancelB)
96− chB, err := b.SubscribeEvents(ctxB, wsB.ID)
97− require.NoError(t, err)
98−
99− // Trigger a republish on workspace A only. The marker name lets us
100− // distinguish this event from any incidental backend activity.
101− const marker = "wsa-republish-marker"
102− wsA.Skills.PublishStates([]*skills.SkillState{
103− {Name: marker, State: skills.StateNormal},
104− })
105−
106− // Workspace A must receive its own event.
107− require.True(t,
108− waitForSkillsEvent(t, chA, marker, 2*time.Second),
109− "workspace A never received its own skills event")
110−
111− // Workspace B must NOT receive workspace A's event.
112− require.False(t,
113− waitForSkillsEvent(t, chB, marker, 250*time.Millisecond),
114− "workspace B leaked workspace A's skills event")
115−
116− // And A's published states are now visible on its manager's
117− // snapshot (verifies PublishStates updates the cache, not just the
118− // broker).
119− updatedA := wsA.Skills.States()
120− require.True(t, containsSkillName(updatedA, marker),
121− "PublishStates must update Manager.States()")
23+// newTestBackend returns a Backend whose teardown path skips any
24+// real [app.App] shutdown work. Useful for state-machine tests that
25+// install synthetic workspaces directly via insertTestWorkspace.
26+func newTestBackend(t *testing.T) (*Backend, *atomic.Int32) {
27+ t.Helper()
28+ var shutdownCount atomic.Int32
29+ b := &Backend{
30+ workspaces: csync.NewMap[string, *Workspace](),
31+ pathIndex: make(map[string]string),
32+ ctx: context.Background(),
33+ createGrace: 50 * time.Millisecond,
34+ shutdownFn: func() { shutdownCount.Add(1) },
35+ }
36+ return b, &shutdownCount
37+}
12238
123− // B's manager snapshot is untouched.
124− require.False(t, containsSkillName(wsB.Skills.States(), marker),
125− "workspace B's Manager.States() leaked workspace A's republish")
39+// insertTestWorkspace installs a synthetic workspace into b at the
40+// given resolved path. Its shutdownFn is recorded in the returned
41+// counter so tests can assert it ran exactly once.
42+func insertTestWorkspace(t *testing.T, b *Backend, key string) (*Workspace, *atomic.Int32) {
43+ t.Helper()
44+ var shutdowns atomic.Int32
45+ ws := &Workspace{
46+ ID: uuid.New().String(),
47+ Path: key,
48+ resolvedPath: key,
49+ clients: make(map[string]*clientState),
50+ shutdownFn: func() { shutdowns.Add(1) },
51+ }
52+ b.mu.Lock()
53+ b.workspaces.Set(ws.ID, ws)
54+ b.pathIndex[key] = ws.ID
55+ b.mu.Unlock()
56+ return ws, &shutdowns
12657}
12758
128−func writeSkill(t *testing.T, workingDir, name, desc string) {
59+func newClientID(t *testing.T) string {
12960 t.Helper()
130− skillDir := filepath.Join(workingDir, ".agents", "skills", name)
131− require.NoError(t, os.MkdirAll(skillDir, 0o755))
132− content := fmt.Sprintf("---\nname: %s\ndescription: %s\n---\n%s\n", name, desc, desc)
133− require.NoError(t, os.WriteFile(filepath.Join(skillDir, "SKILL.md"), []byte(content), 0o644))
61+ return uuid.New().String()
62+}
63+
64+func TestResolveWorkspaceKey_AbsoluteAndSymlink(t *testing.T) {
65+ t.Parallel()
66+
67+ tmp := t.TempDir()
68+ real, err := filepath.EvalSymlinks(tmp)
69+ require.NoError(t, err)
70+
71+ got, err := resolveWorkspaceKey(tmp)
72+ require.NoError(t, err)
73+ require.Equal(t, real, got)
74+}
75+
76+func TestResolveWorkspaceKey_NonExistentFallback(t *testing.T) {
77+ t.Parallel()
78+
79+ missing := filepath.Join(t.TempDir(), "does", "not", "exist")
80+ got, err := resolveWorkspaceKey(missing)
81+ require.NoError(t, err)
82+ abs, err := filepath.Abs(missing)
83+ require.NoError(t, err)
84+ require.Equal(t, abs, got)
85+}
86+
87+func TestValidateClientID(t *testing.T) {
88+ t.Parallel()
89+
90+ _, err := validateClientID("")
91+ require.ErrorIs(t, err, ErrInvalidClientID)
92+ _, err = validateClientID("not-a-uuid")
93+ require.ErrorIs(t, err, ErrInvalidClientID)
94+
95+ id := uuid.New().String()
96+ got, err := validateClientID(id)
97+ require.NoError(t, err)
98+ require.Equal(t, id, got)
99+}
100+
101+func TestRegisterClient_Idempotent(t *testing.T) {
102+ t.Parallel()
103+
104+ b, _ := newTestBackend(t)
105+ ws, _ := insertTestWorkspace(t, b, "/tmp/a")
106+
107+ cid := newClientID(t)
108+ b.registerClient(ws, cid)
109+ b.registerClient(ws, cid)
110+
111+ ws.clientsMu.Lock()
112+ defer ws.clientsMu.Unlock()
113+ require.Len(t, ws.clients, 1)
114+ require.NotNil(t, ws.clients[cid].holdTimer)
115+ require.Equal(t, 0, ws.clients[cid].streams)
116+}
117+
118+func TestAttachClient_ConsumesHold(t *testing.T) {
119+ t.Parallel()
120+
121+ b, _ := newTestBackend(t)
122+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
123+
124+ cid := newClientID(t)
125+ b.registerClient(ws, cid)
126+ require.NoError(t, b.AttachClient(ws.ID, cid))
127+
128+ ws.clientsMu.Lock()
129+ require.Len(t, ws.clients, 1)
130+ require.Nil(t, ws.clients[cid].holdTimer, "attach must stop the grace timer")
131+ require.Equal(t, 1, ws.clients[cid].streams)
132+ ws.clientsMu.Unlock()
133+
134+ // Wait past the grace window: a stopped timer must not fire.
135+ time.Sleep(150 * time.Millisecond)
136+ require.Equal(t, int32(0), shutdowns.Load(), "workspace must not be torn down while attached")
137+}
138+
139+func TestAttachClient_WithoutPriorCreate(t *testing.T) {
140+ t.Parallel()
141+
142+ b, _ := newTestBackend(t)
143+ ws, _ := insertTestWorkspace(t, b, "/tmp/a")
144+
145+ cid := newClientID(t)
146+ require.NoError(t, b.AttachClient(ws.ID, cid))
147+
148+ ws.clientsMu.Lock()
149+ defer ws.clientsMu.Unlock()
150+ require.Len(t, ws.clients, 1)
151+ require.Equal(t, 1, ws.clients[cid].streams)
152+ require.Nil(t, ws.clients[cid].holdTimer)
153+}
154+
155+func TestAttachClient_DuplicateStreams(t *testing.T) {
156+ t.Parallel()
157+
158+ b, _ := newTestBackend(t)
159+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
160+
161+ cid := newClientID(t)
162+ require.NoError(t, b.AttachClient(ws.ID, cid))
163+ require.NoError(t, b.AttachClient(ws.ID, cid))
164+
165+ ws.clientsMu.Lock()
166+ require.Equal(t, 2, ws.clients[cid].streams)
167+ ws.clientsMu.Unlock()
168+
169+ b.DetachClient(ws.ID, cid)
170+ ws.clientsMu.Lock()
171+ require.Equal(t, 1, ws.clients[cid].streams)
172+ ws.clientsMu.Unlock()
173+ require.Equal(t, int32(0), shutdowns.Load())
174+
175+ b.DetachClient(ws.ID, cid)
176+ require.Equal(t, int32(1), shutdowns.Load(), "second detach tears down the workspace")
134177}
135178
136−func containsSkillName(states []*skills.SkillState, name string) bool {
137− for _, s := range states {
138− if s.Name == name {
139− return true
179+func TestDetachClient_LastStreamTearsDown(t *testing.T) {
180+ t.Parallel()
181+
182+ b, srvShutdowns := newTestBackend(t)
183+ ws, wsShutdowns := insertTestWorkspace(t, b, "/tmp/a")
184+
185+ cid := newClientID(t)
186+ b.registerClient(ws, cid)
187+ require.NoError(t, b.AttachClient(ws.ID, cid))
188+ b.DetachClient(ws.ID, cid)
189+
190+ require.Equal(t, int32(1), wsShutdowns.Load())
191+ require.Equal(t, int32(1), srvShutdowns.Load(), "last workspace shut down must trigger server shutdown")
192+ _, err := b.GetWorkspace(ws.ID)
193+ require.ErrorIs(t, err, ErrWorkspaceNotFound)
194+}
195+
196+func TestHoldExpiry_TearsDown(t *testing.T) {
197+ t.Parallel()
198+
199+ b, srvShutdowns := newTestBackend(t)
200+ ws, wsShutdowns := insertTestWorkspace(t, b, "/tmp/a")
201+
202+ cid := newClientID(t)
203+ b.registerClient(ws, cid)
204+
205+ require.Eventually(t, func() bool {
206+ return wsShutdowns.Load() == 1 && srvShutdowns.Load() == 1
207+ }, 1*time.Second, 5*time.Millisecond)
208+}
209+
210+func TestReleaseHold_NoStreams(t *testing.T) {
211+ t.Parallel()
212+
213+ b, _ := newTestBackend(t)
214+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
215+
216+ cid := newClientID(t)
217+ b.registerClient(ws, cid)
218+ require.NoError(t, b.releaseHold(ws.ID, cid))
219+
220+ require.Equal(t, int32(1), shutdowns.Load())
221+ // Idempotent.
222+ require.NoError(t, b.releaseHold(ws.ID, cid))
223+ require.Equal(t, int32(1), shutdowns.Load())
224+}
225+
226+func TestReleaseHold_WithActiveStream(t *testing.T) {
227+ t.Parallel()
228+
229+ b, _ := newTestBackend(t)
230+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
231+
232+ cid := newClientID(t)
233+ b.registerClient(ws, cid)
234+ require.NoError(t, b.AttachClient(ws.ID, cid))
235+ require.NoError(t, b.releaseHold(ws.ID, cid))
236+
237+ ws.clientsMu.Lock()
238+ require.Equal(t, 1, ws.clients[cid].streams)
239+ require.Nil(t, ws.clients[cid].holdTimer)
240+ ws.clientsMu.Unlock()
241+ require.Equal(t, int32(0), shutdowns.Load())
242+
243+ b.DetachClient(ws.ID, cid)
244+ require.Equal(t, int32(1), shutdowns.Load())
245+}
246+
247+func TestReleaseHoldThenAttach(t *testing.T) {
248+ t.Parallel()
249+
250+ b, _ := newTestBackend(t)
251+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
252+
253+ cid := newClientID(t)
254+ require.NoError(t, b.releaseHold(ws.ID, cid)) // no entry yet — no-op.
255+ require.NoError(t, b.AttachClient(ws.ID, cid))
256+ ws.clientsMu.Lock()
257+ require.Equal(t, 1, ws.clients[cid].streams)
258+ ws.clientsMu.Unlock()
259+ require.NoError(t, b.releaseHold(ws.ID, cid)) // hold-only no-op (no hold timer).
260+ require.Equal(t, int32(0), shutdowns.Load())
261+ b.DetachClient(ws.ID, cid)
262+ require.Equal(t, int32(1), shutdowns.Load())
263+}
264+
265+func TestRefcountWithSecondClient(t *testing.T) {
266+ t.Parallel()
267+
268+ b, _ := newTestBackend(t)
269+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/a")
270+
271+ cidA := newClientID(t)
272+ cidB := newClientID(t)
273+ b.registerClient(ws, cidA)
274+ require.NoError(t, b.AttachClient(ws.ID, cidA))
275+ b.registerClient(ws, cidB)
276+ require.NoError(t, b.AttachClient(ws.ID, cidB))
277+
278+ b.DetachClient(ws.ID, cidA)
279+ ws.clientsMu.Lock()
280+ require.Contains(t, ws.clients, cidB)
281+ require.NotContains(t, ws.clients, cidA)
282+ ws.clientsMu.Unlock()
283+ require.Equal(t, int32(0), shutdowns.Load(), "workspace survives while second client attached")
284+
285+ b.DetachClient(ws.ID, cidB)
286+ require.Equal(t, int32(1), shutdowns.Load())
287+}
288+
289+func TestAttachClient_InvalidID(t *testing.T) {
290+ t.Parallel()
291+
292+ b, _ := newTestBackend(t)
293+ ws, _ := insertTestWorkspace(t, b, "/tmp/a")
294+
295+ require.ErrorIs(t, b.AttachClient(ws.ID, ""), ErrInvalidClientID)
296+ require.ErrorIs(t, b.AttachClient(ws.ID, "not-a-uuid"), ErrInvalidClientID)
297+}
298+
299+func TestDeleteWorkspace_RejectsBadClientID(t *testing.T) {
300+ t.Parallel()
301+
302+ b, _ := newTestBackend(t)
303+ ws, _ := insertTestWorkspace(t, b, "/tmp/a")
304+
305+ require.ErrorIs(t, b.DeleteWorkspace(ws.ID, ""), ErrInvalidClientID)
306+ require.ErrorIs(t, b.DeleteWorkspace(ws.ID, "not-a-uuid"), ErrInvalidClientID)
307+}
308+
309+// TestHoldExpiry_RaceWithAttach checks that, when the grace timer fires
310+// while a concurrent AttachClient call is in flight, the workspace ends
311+// up either fully attached or fully torn down — never in a half-state.
312+func TestHoldExpiry_RaceWithAttach(t *testing.T) {
313+ t.Parallel()
314+
315+ for i := range 50 {
316+ b, _ := newTestBackend(t)
317+ // Tighten the grace window further to force the race.
318+ b.createGrace = 1 * time.Millisecond
319+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/race")
320+
321+ cid := newClientID(t)
322+ b.registerClient(ws, cid)
323+ // Attach concurrently with the very short grace timer.
324+ errCh := make(chan error, 1)
325+ go func() { errCh <- b.AttachClient(ws.ID, cid) }()
326+ <-errCh
327+
328+ // Wait for any pending timer to settle.
329+ time.Sleep(10 * time.Millisecond)
330+
331+ ws.clientsMu.Lock()
332+ gotShutdown := shutdowns.Load() == 1
333+ cs, present := ws.clients[cid]
334+ var (
335+ gotStreams int
336+ gotHoldTimer *time.Timer
337+ )
338+ if present {
339+ gotStreams = cs.streams
340+ gotHoldTimer = cs.holdTimer
341+ }
342+ ws.clientsMu.Unlock()
343+ // Either the workspace was torn down OR the client is
344+ // attached with streams==1 and the hold timer cleared.
345+ // The state must be consistent: if shutdown, client is
346+ // gone; if attached, no teardown and streams==1.
347+ if gotShutdown {
348+ require.False(t, present, "iter %d: shutdown but client still present", i)
349+ } else {
350+ require.True(t, present, "iter %d: not shutdown but client missing", i)
351+ require.Equal(t, 1, gotStreams, "iter %d: attach winner must leave streams=1", i)
352+ require.Nil(t, gotHoldTimer, "iter %d: attach winner must clear holdTimer", i)
140353 }
141354 }
142− return false
143355}
144356
145−// waitForSkillsEvent drains the given event channel until either a
146−// pubsub.Event[skills.Event] containing a state named marker arrives or
147−// the timeout elapses. Non-skills events on the channel are silently
148−// skipped — the backend fans in many event types and we only care
149−// about skills here.
150−func waitForSkillsEvent(t *testing.T, ch <-chan pubsub.Event[tea.Msg], marker string, timeout time.Duration) bool {
357+// TestConcurrentAttachDetach exercises the state machine under
358+// parallel attach/detach pressure with the race detector.
359+func TestConcurrentAttachDetach(t *testing.T) {
360+ t.Parallel()
361+
362+ b, _ := newTestBackend(t)
363+ ws, _ := insertTestWorkspace(t, b, "/tmp/a")
364+
365+ cid := newClientID(t)
366+ b.registerClient(ws, cid)
367+ require.NoError(t, b.AttachClient(ws.ID, cid)) // ensure refcount stays > 0.
368+
369+ const n = 50
370+ var wg sync.WaitGroup
371+ wg.Add(n)
372+ for range n {
373+ go func() {
374+ defer wg.Done()
375+ cid2 := newClientID(t)
376+ _ = b.AttachClient(ws.ID, cid2)
377+ b.DetachClient(ws.ID, cid2)
378+ }()
379+ }
380+ wg.Wait()
381+
382+ ws.clientsMu.Lock()
383+ defer ws.clientsMu.Unlock()
384+ require.Len(t, ws.clients, 1)
385+ require.Contains(t, ws.clients, cid)
386+}
387+
388+// TestPathDedupe_FullCreate exercises CreateWorkspace end-to-end
389+// (config init, real app.App). Two CreateWorkspace calls at the same
390+// path return the same workspace ID and share the clients map.
391+func TestPathDedupe_FullCreate(t *testing.T) {
392+ t.Setenv("HOME", t.TempDir())
393+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
394+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
395+ t.Setenv("XDG_DATA_HOME", t.TempDir())
396+
397+ cwd := t.TempDir()
398+ dataDir := t.TempDir()
399+
400+ b := New(context.Background(), nil, func() {})
401+ b.SetCreateGrace(2 * time.Second)
402+ t.Cleanup(func() { drainBackend(t, b) })
403+
404+ cidA := uuid.New().String()
405+ cidB := uuid.New().String()
406+
407+ wsA, protoA, err := b.CreateWorkspace(protoWS(cwd, dataDir, cidA))
408+ require.NoError(t, err)
409+ require.NotEmpty(t, protoA.ID)
410+ require.Equal(t, protoA.DataDir, wsA.Cfg.Config().Options.DataDirectory)
411+
412+ wsB, protoB, err := b.CreateWorkspace(protoWS(cwd, dataDir, cidB))
413+ require.NoError(t, err)
414+ require.Equal(t, wsA.ID, wsB.ID, "second create at same path must return existing workspace")
415+ require.Equal(t, protoA.ID, protoB.ID)
416+
417+ wsA.clientsMu.Lock()
418+ require.Contains(t, wsA.clients, cidA)
419+ require.Contains(t, wsA.clients, cidB)
420+ wsA.clientsMu.Unlock()
421+}
422+
423+// TestPathDedupe_DifferentPaths_DifferentWorkspaces confirms that two
424+// CreateWorkspace calls at distinct paths produce distinct workspaces.
425+func TestPathDedupe_DifferentPaths_DifferentWorkspaces(t *testing.T) {
426+ t.Setenv("HOME", t.TempDir())
427+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
428+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
429+ t.Setenv("XDG_DATA_HOME", t.TempDir())
430+
431+ cwdA := t.TempDir()
432+ cwdB := t.TempDir()
433+ dataA := t.TempDir()
434+ dataB := t.TempDir()
435+
436+ b := New(context.Background(), nil, func() {})
437+ b.SetCreateGrace(2 * time.Second)
438+ t.Cleanup(func() { drainBackend(t, b) })
439+
440+ wsA, _, err := b.CreateWorkspace(protoWS(cwdA, dataA, uuid.New().String()))
441+ require.NoError(t, err)
442+ wsB, _, err := b.CreateWorkspace(protoWS(cwdB, dataB, uuid.New().String()))
443+ require.NoError(t, err)
444+ require.NotEqual(t, wsA.ID, wsB.ID)
445+}
446+
447+// TestPathDedupe_FirstWinsKeepsOriginalEnv verifies that the second
448+// create at the same path returns the *originating* client's Env in
449+// its proto and does not mutate the existing workspace's YOLO/Debug
450+// flags.
451+func TestPathDedupe_FirstWinsKeepsOriginalEnv(t *testing.T) {
452+ t.Setenv("HOME", t.TempDir())
453+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
454+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
455+ t.Setenv("XDG_DATA_HOME", t.TempDir())
456+
457+ cwd := t.TempDir()
458+ dataDir := t.TempDir()
459+
460+ b := New(context.Background(), nil, func() {})
461+ b.SetCreateGrace(2 * time.Second)
462+ t.Cleanup(func() { drainBackend(t, b) })
463+
464+ originalEnv := []string{"FOO=bar"}
465+ argsA := protoWS(cwd, dataDir, uuid.New().String())
466+ argsA.YOLO = true
467+ argsA.Env = originalEnv
468+ wsA, protoA, err := b.CreateWorkspace(argsA)
469+ require.NoError(t, err)
470+ require.True(t, protoA.YOLO)
471+ require.Equal(t, originalEnv, protoA.Env)
472+
473+ argsB := protoWS(cwd, dataDir, uuid.New().String())
474+ argsB.YOLO = false
475+ argsB.Debug = true
476+ argsB.Env = []string{"BAZ=qux"}
477+ _, protoB, err := b.CreateWorkspace(argsB)
478+ require.NoError(t, err)
479+ require.Equal(t, protoA.ID, protoB.ID)
480+ require.True(t, protoB.YOLO, "first wins: YOLO must remain true")
481+ require.Equal(t, originalEnv, protoB.Env, "proto must carry the originating client's Env")
482+ require.Equal(t, wsA.Cfg.Overrides().SkipPermissionRequests, true)
483+}
484+
485+// TestPathDedupe_Symlink confirms two paths that resolve to the same
486+// target share a workspace.
487+func TestPathDedupe_Symlink(t *testing.T) {
488+ if runtime.GOOS == "windows" {
489+ t.Skip("symlink semantics differ on windows")
490+ }
491+ t.Setenv("HOME", t.TempDir())
492+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
493+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
494+ t.Setenv("XDG_DATA_HOME", t.TempDir())
495+
496+ real := t.TempDir()
497+ link := filepath.Join(t.TempDir(), "link")
498+ require.NoError(t, os.Symlink(real, link))
499+ dataDir := t.TempDir()
500+
501+ b := New(context.Background(), nil, func() {})
502+ b.SetCreateGrace(2 * time.Second)
503+ t.Cleanup(func() { drainBackend(t, b) })
504+
505+ wsA, _, err := b.CreateWorkspace(protoWS(real, dataDir, uuid.New().String()))
506+ require.NoError(t, err)
507+ wsB, _, err := b.CreateWorkspace(protoWS(link, dataDir, uuid.New().String()))
508+ require.NoError(t, err)
509+ require.Equal(t, wsA.ID, wsB.ID)
510+}
511+
512+// TestPathDedupe_NonExistentPath ensures CreateWorkspace tolerates a
513+// path that does not yet exist (EvalSymlinks falls back to Abs).
514+func TestPathDedupe_NonExistentPath(t *testing.T) {
515+ t.Setenv("HOME", t.TempDir())
516+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
517+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
518+ t.Setenv("XDG_DATA_HOME", t.TempDir())
519+
520+ parent := t.TempDir()
521+ missing := filepath.Join(parent, "does-not-exist")
522+ dataDir := t.TempDir()
523+
524+ b := New(context.Background(), nil, func() {})
525+ b.SetCreateGrace(2 * time.Second)
526+ t.Cleanup(func() { drainBackend(t, b) })
527+
528+ _, p, err := b.CreateWorkspace(protoWS(missing, dataDir, uuid.New().String()))
529+ require.NoError(t, err)
530+ require.NotEmpty(t, p.ID)
531+}
532+
533+// TestCreateWorkspace_IdempotentSameClient checks that a duplicate
534+// create from the same client at the same path does not produce a
535+// second claim.
536+func TestCreateWorkspace_IdempotentSameClient(t *testing.T) {
537+ t.Setenv("HOME", t.TempDir())
538+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
539+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
540+ t.Setenv("XDG_DATA_HOME", t.TempDir())
541+
542+ cwd := t.TempDir()
543+ dataDir := t.TempDir()
544+ b := New(context.Background(), nil, func() {})
545+ b.SetCreateGrace(2 * time.Second)
546+ t.Cleanup(func() { drainBackend(t, b) })
547+
548+ cid := uuid.New().String()
549+ ws1, _, err := b.CreateWorkspace(protoWS(cwd, dataDir, cid))
550+ require.NoError(t, err)
551+ ws2, _, err := b.CreateWorkspace(protoWS(cwd, dataDir, cid))
552+ require.NoError(t, err)
553+ require.Equal(t, ws1.ID, ws2.ID)
554+
555+ ws1.clientsMu.Lock()
556+ require.Len(t, ws1.clients, 1, "duplicate create from same client must not double the claim")
557+ ws1.clientsMu.Unlock()
558+}
559+
560+// TestPathDedupe_ParallelCreates ensures two simultaneous CreateWorkspace
561+// calls at the same path produce the same workspace and the clients map
562+// contains both client IDs.
563+func TestPathDedupe_ParallelCreates(t *testing.T) {
564+ t.Setenv("HOME", t.TempDir())
565+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
566+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
567+ t.Setenv("XDG_DATA_HOME", t.TempDir())
568+
569+ cwd := t.TempDir()
570+ dataDir := t.TempDir()
571+
572+ b := New(context.Background(), nil, func() {})
573+ b.SetCreateGrace(2 * time.Second)
574+ t.Cleanup(func() { drainBackend(t, b) })
575+
576+ cidA := uuid.New().String()
577+ cidB := uuid.New().String()
578+
579+ type result struct {
580+ ws *Workspace
581+ proto proto.Workspace
582+ err error
583+ }
584+ ch := make(chan result, 2)
585+ start := make(chan struct{})
586+ go func() {
587+ <-start
588+ ws, p, err := b.CreateWorkspace(protoWS(cwd, dataDir, cidA))
589+ ch <- result{ws, p, err}
590+ }()
591+ go func() {
592+ <-start
593+ ws, p, err := b.CreateWorkspace(protoWS(cwd, dataDir, cidB))
594+ ch <- result{ws, p, err}
595+ }()
596+ close(start)
597+ r1 := <-ch
598+ r2 := <-ch
599+ require.NoError(t, r1.err)
600+ require.NoError(t, r2.err)
601+ require.Equal(t, r1.ws.ID, r2.ws.ID, "both creates must converge on one workspace ID")
602+
603+ ws := r1.ws
604+ ws.clientsMu.Lock()
605+ defer ws.clientsMu.Unlock()
606+ require.Contains(t, ws.clients, cidA)
607+ require.Contains(t, ws.clients, cidB)
608+}
609+
610+// TestCreateWorkspace_RejectsBadClientID covers the 400 path from the
611+// backend side.
612+func TestCreateWorkspace_RejectsBadClientID(t *testing.T) {
613+ t.Parallel()
614+
615+ b := New(context.Background(), nil, func() {})
616+
617+ _, _, err := b.CreateWorkspace(protoWS("/tmp/x", t.TempDir(), ""))
618+ require.ErrorIs(t, err, ErrInvalidClientID)
619+ _, _, err = b.CreateWorkspace(protoWS("/tmp/x", t.TempDir(), "not-a-uuid"))
620+ require.ErrorIs(t, err, ErrInvalidClientID)
621+}
622+
623+// drainBackend tears the backend down at the end of a test by deleting
624+// every remaining workspace. Necessary so the test process doesn't
625+// leak goroutines or DB handles from the embedded [app.App] instances.
626+func drainBackend(t *testing.T, b *Backend) {
151627 t.Helper()
152− deadline := time.After(timeout)
153− for {
154− select {
155− case ev, ok := <-ch:
156− if !ok {
157− return false
158− }
159− se, ok := ev.Payload.(pubsub.Event[skills.Event])
160− if !ok {
161− continue
162− }
163− if containsSkillName(se.Payload.States, marker) {
164− return true
165− }
166− case <-deadline:
167− return false
628+ for _, ws := range b.workspaces.Seq2() {
629+ ws.clientsMu.Lock()
630+ ids := make([]string, 0, len(ws.clients))
631+ for id := range ws.clients {
632+ ids = append(ids, id)
633+ }
634+ ws.clientsMu.Unlock()
635+ for _, cid := range ids {
636+ _ = b.releaseHold(ws.ID, cid)
637+ }
638+ }
639+}
640+
641+func protoWS(path, dataDir, clientID string) proto.Workspace {
642+ return proto.Workspace{Path: path, DataDir: dataDir, ClientID: clientID}
643+}
644+
645+// syncBuffer is a thread-safe buffer that can be safely read and written
646+// from multiple goroutines.
647+type syncBuffer struct {
648+ mu sync.Mutex
649+ buf bytes.Buffer
650+}
651+
652+func (sb *syncBuffer) Write(p []byte) (n int, err error) {
653+ sb.mu.Lock()
654+ defer sb.mu.Unlock()
655+ return sb.buf.Write(p)
656+}
657+
658+func (sb *syncBuffer) String() string {
659+ sb.mu.Lock()
660+ defer sb.mu.Unlock()
661+ return sb.buf.String()
662+}
663+
664+// captureDebugLogs installs a buffer-backed slog handler at Debug
665+// level for the duration of the test, returning the buffer. The
666+// previous default handler is restored via t.Cleanup.
667+func captureDebugLogs(t *testing.T) *syncBuffer {
668+ t.Helper()
669+ var sb syncBuffer
670+ prev := slog.Default()
671+ handler := slog.NewTextHandler(&sb, &slog.HandlerOptions{Level: slog.LevelDebug})
672+ slog.SetDefault(slog.New(handler))
673+ t.Cleanup(func() { slog.SetDefault(prev) })
674+ return &sb
675+}
676+
677+// xdgIsolated points HOME and XDG_* variables at fresh tempdirs so
678+// CreateWorkspace's config loading does not interfere with the host
679+// machine's real config.
680+func xdgIsolated(t *testing.T) {
681+ t.Helper()
682+ t.Setenv("HOME", t.TempDir())
683+ t.Setenv("XDG_CACHE_HOME", t.TempDir())
684+ t.Setenv("XDG_CONFIG_HOME", t.TempDir())
685+ t.Setenv("XDG_DATA_HOME", t.TempDir())
686+}
687+
688+// TestFirstWinsMismatch_LogsOnFlagDifferences verifies that the
689+// debug mismatch line is emitted when any of YOLO, Debug, DataDir,
690+// or Env differs between the first and second CreateWorkspace at
691+// the same path, and that the existing workspace's Debug flag is
692+// not overwritten.
693+func TestFirstWinsMismatch_LogsOnFlagDifferences(t *testing.T) {
694+ tests := []struct {
695+ name string
696+ mutate func(*proto.Workspace)
697+ }{
698+ {
699+ name: "yolo",
700+ mutate: func(p *proto.Workspace) { p.YOLO = true },
701+ },
702+ {
703+ name: "debug",
704+ mutate: func(p *proto.Workspace) { p.Debug = true },
705+ },
706+ {
707+ name: "datadir",
708+ mutate: func(p *proto.Workspace) { p.DataDir = "" },
709+ },
710+ {
711+ name: "env",
712+ mutate: func(p *proto.Workspace) { p.Env = []string{"NEW=val"} },
713+ },
714+ }
715+
716+ for _, tc := range tests {
717+ t.Run(tc.name, func(t *testing.T) {
718+ xdgIsolated(t)
719+ cwd := t.TempDir()
720+ dataDir := t.TempDir()
721+
722+ buf := captureDebugLogs(t)
723+ b := New(context.Background(), nil, func() {})
724+ b.SetCreateGrace(2 * time.Second)
725+ t.Cleanup(func() { drainBackend(t, b) })
726+
727+ argsA := protoWS(cwd, dataDir, uuid.New().String())
728+ argsA.Env = []string{"FOO=bar"}
729+ wsA, _, err := b.CreateWorkspace(argsA)
730+ require.NoError(t, err)
731+ originalDebug := wsA.Cfg.Config().Options.Debug
732+ originalYOLO := wsA.Cfg.Overrides().SkipPermissionRequests
733+
734+ argsB := protoWS(cwd, dataDir, uuid.New().String())
735+ argsB.Env = []string{"FOO=bar"} // identical by default
736+ tc.mutate(&argsB)
737+ _, _, err = b.CreateWorkspace(argsB)
738+ require.NoError(t, err)
739+
740+ require.Contains(
741+ t, buf.String(),
742+ "Workspace flag mismatch on duplicate create",
743+ "expected debug log for mismatching %s", tc.name,
744+ )
745+ // Existing workspace's YOLO and Debug must not change.
746+ require.Equal(t, originalYOLO, wsA.Cfg.Overrides().SkipPermissionRequests, "YOLO must be immutable on first-wins")
747+ require.Equal(t, originalDebug, wsA.Cfg.Config().Options.Debug, "Debug must be immutable on first-wins")
748+ })
749+ }
750+}
751+
752+// TestFirstWinsMismatch_NoLogWhenIdentical confirms identical args
753+// do not emit the mismatch log line.
754+func TestFirstWinsMismatch_NoLogWhenIdentical(t *testing.T) {
755+ xdgIsolated(t)
756+ cwd := t.TempDir()
757+ dataDir := t.TempDir()
758+
759+ buf := captureDebugLogs(t)
760+ b := New(context.Background(), nil, func() {})
761+ b.SetCreateGrace(2 * time.Second)
762+ t.Cleanup(func() { drainBackend(t, b) })
763+
764+ argsA := protoWS(cwd, dataDir, uuid.New().String())
765+ argsA.Env = []string{"FOO=bar"}
766+ _, _, err := b.CreateWorkspace(argsA)
767+ require.NoError(t, err)
768+
769+ argsB := protoWS(cwd, dataDir, uuid.New().String())
770+ argsB.Env = []string{"FOO=bar"}
771+ _, _, err = b.CreateWorkspace(argsB)
772+ require.NoError(t, err)
773+
774+ require.False(t,
775+ strings.Contains(buf.String(), "Workspace flag mismatch on duplicate create"),
776+ "identical args must not log a mismatch: %s", buf.String())
777+}
778+
779+// TestRaceTwoClientsAttachOneDetaches exercises the PLAN-required
780+// race scenario: two clients attach concurrently, then one detaches.
781+// The workspace must remain alive with refcount==1 and the clients
782+// map must reflect the remaining client only.
783+func TestRaceTwoClientsAttachOneDetaches(t *testing.T) {
784+ t.Parallel()
785+
786+ b, _ := newTestBackend(t)
787+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/race-two")
788+
789+ cidA := newClientID(t)
790+ cidB := newClientID(t)
791+
792+ var wg sync.WaitGroup
793+ wg.Add(2)
794+ go func() {
795+ defer wg.Done()
796+ require.NoError(t, b.AttachClient(ws.ID, cidA))
797+ }()
798+ go func() {
799+ defer wg.Done()
800+ require.NoError(t, b.AttachClient(ws.ID, cidB))
801+ }()
802+ wg.Wait()
803+
804+ ws.clientsMu.Lock()
805+ require.Len(t, ws.clients, 2, "both clients must be attached")
806+ ws.clientsMu.Unlock()
807+
808+ b.DetachClient(ws.ID, cidA)
809+
810+ ws.clientsMu.Lock()
811+ require.Len(t, ws.clients, 1, "refcount must be 1 after one detach")
812+ require.Contains(t, ws.clients, cidB, "remaining client must be cidB")
813+ require.NotContains(t, ws.clients, cidA, "detached client must be removed")
814+ ws.clientsMu.Unlock()
815+ require.Equal(t, int32(0), shutdowns.Load(), "workspace must remain alive")
816+
817+ // Drain.
818+ b.DetachClient(ws.ID, cidB)
819+ require.Equal(t, int32(1), shutdowns.Load())
820+}
821+
822+// TestExplicitDeleteThenAttach reproduces the PLAN scenario: start
823+// with a real hold, releaseHold consumes it, AttachClient from the
824+// same clientID creates a fresh entry with streams==1, and calling
825+// releaseHold again is a no-op. A second client keeps the workspace
826+// alive so AttachClient can still resolve the workspace ID after the
827+// first client's hold is released.
828+func TestExplicitDeleteThenAttach(t *testing.T) {
829+ t.Parallel()
830+
831+ // Large grace window so timers cannot fire during the test
832+ // — we want to exercise the explicit releaseHold path.
833+ b, _ := newTestBackend(t)
834+ b.createGrace = time.Hour
835+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/delete-then-attach")
836+
837+ // Anchor client keeps the workspace registered in
838+ // b.workspaces across the cid's releaseHold below.
839+ anchor := newClientID(t)
840+ require.NoError(t, b.AttachClient(ws.ID, anchor))
841+
842+ cid := newClientID(t)
843+ // Real hold via registerClient (mirrors CreateWorkspace).
844+ b.registerClient(ws, cid)
845+ ws.clientsMu.Lock()
846+ require.Contains(t, ws.clients, cid)
847+ require.NotNil(t, ws.clients[cid].holdTimer, "hold must be live")
848+ require.Equal(t, 0, ws.clients[cid].streams)
849+ ws.clientsMu.Unlock()
850+
851+ // releaseHold: consumes the hold and removes the entry
852+ // (streams == 0). The anchor client keeps the workspace
853+ // alive.
854+ require.NoError(t, b.releaseHold(ws.ID, cid))
855+ require.Equal(t, int32(0), shutdowns.Load(), "anchor must keep workspace alive")
856+ ws.clientsMu.Lock()
857+ require.NotContains(t, ws.clients, cid, "entry must be removed by releaseHold")
858+ ws.clientsMu.Unlock()
859+
860+ // AttachClient creates a fresh entry with streams==1 and no
861+ // hold timer.
862+ require.NoError(t, b.AttachClient(ws.ID, cid))
863+ ws.clientsMu.Lock()
864+ require.Contains(t, ws.clients, cid, "fresh entry must be created")
865+ require.Equal(t, 1, ws.clients[cid].streams, "fresh attach must start at streams=1")
866+ require.Nil(t, ws.clients[cid].holdTimer, "fresh attach must have no hold timer")
867+ ws.clientsMu.Unlock()
868+
869+ // Calling releaseHold again is a no-op (no hold timer to
870+ // stop, streams > 0 so the entry stays).
871+ require.NoError(t, b.releaseHold(ws.ID, cid))
872+ ws.clientsMu.Lock()
873+ require.Contains(t, ws.clients, cid, "releaseHold must not touch a stream-only entry")
874+ require.Equal(t, 1, ws.clients[cid].streams)
875+ require.Nil(t, ws.clients[cid].holdTimer)
876+ ws.clientsMu.Unlock()
877+
878+ // Drain.
879+ b.DetachClient(ws.ID, cid)
880+ b.DetachClient(ws.ID, anchor)
881+ require.Equal(t, int32(1), shutdowns.Load())
882+}
883+
884+// TestAttachClient_RacesWithTeardown forces AttachClient to compete
885+// with the teardown path triggered by DetachClient. Before the fix,
886+// AttachClient could observe a workspace after teardown had already
887+// decided to remove it (because AttachClient did not synchronize with
888+// Backend.mu), leaving a live stream claim attached to a workspace
889+// that was then removed and shut down. With the fix, the outcome must
890+// be deterministic: either AttachClient won and the workspace is
891+// alive with the client registered, or teardown won and AttachClient
892+// returns ErrWorkspaceNotFound — never a half-state where the
893+// workspace is gone but ws.clients still contains the new client.
894+func TestAttachClient_RacesWithTeardown(t *testing.T) {
895+ t.Parallel()
896+
897+ for i := range 200 {
898+ b, _ := newTestBackend(t)
899+ // Keep the grace window long so it can't fire during the
900+ // test and confuse the bookkeeping.
901+ b.createGrace = time.Hour
902+ ws, shutdowns := insertTestWorkspace(t, b, "/tmp/race-teardown")
903+
904+ // Seed: cidA holds the workspace open via a stream. The
905+ // imminent DetachClient(cidA) will be the *only* claim
906+ // drop, so teardown will run.
907+ cidA := newClientID(t)
908+ require.NoError(t, b.AttachClient(ws.ID, cidA))
909+
910+ // cidB attempts to attach concurrently with the detach
911+ // that will tear the workspace down.
912+ cidB := newClientID(t)
913+ start := make(chan struct{})
914+ errCh := make(chan error, 1)
915+ detachDone := make(chan struct{})
916+ go func() {
917+ <-start
918+ errCh <- b.AttachClient(ws.ID, cidB)
919+ }()
920+ go func() {
921+ <-start
922+ b.DetachClient(ws.ID, cidA)
923+ close(detachDone)
924+ }()
925+ close(start)
926+
927+ // Wait for both goroutines so teardown (including
928+ // shutdownFn) has fully run before we read state.
929+ attachErr := <-errCh
930+ <-detachDone
931+
932+ _, wsStillRegistered := b.workspaces.Get(ws.ID)
933+ ws.clientsMu.Lock()
934+ _, hasA := ws.clients[cidA]
935+ _, hasB := ws.clients[cidB]
936+ clientCount := len(ws.clients)
937+ ws.clientsMu.Unlock()
938+ shutdownCount := shutdowns.Load()
939+
940+ switch {
941+ case attachErr == nil:
942+ // AttachClient won. The workspace must be alive
943+ // (registered) with cidB in its clients map. cidA
944+ // may or may not still be there depending on who
945+ // took clientsMu first, but the workspace must
946+ // not have been torn down.
947+ require.True(t, wsStillRegistered,
948+ "iter %d: attach succeeded but workspace was removed", i)
949+ require.True(t, hasB,
950+ "iter %d: attach succeeded but cidB missing from clients", i)
951+ require.Equal(t, int32(0), shutdownCount,
952+ "iter %d: attach succeeded but workspace was shut down", i)
953+ case errors.Is(attachErr, ErrWorkspaceNotFound):
954+ // Teardown won. The workspace must be removed,
955+ // shut down exactly once, and ws.clients must be
956+ // empty (no half-state with cidB inserted into a
957+ // dead workspace's clients map).
958+ require.False(t, wsStillRegistered,
959+ "iter %d: ErrWorkspaceNotFound but workspace still registered", i)
960+ require.Equal(t, int32(1), shutdownCount,
961+ "iter %d: ErrWorkspaceNotFound but shutdown count = %d", i, shutdownCount)
962+ require.False(t, hasA,
963+ "iter %d: teardown won but cidA still in clients", i)
964+ require.False(t, hasB,
965+ "iter %d: teardown won but cidB still in clients (would be the leaked attach)", i)
966+ require.Zero(t, clientCount,
967+ "iter %d: teardown won but clients map is non-empty", i)
968+ default:
969+ t.Fatalf("iter %d: unexpected AttachClient error: %v", i, attachErr)
168970 }
169971 }
170972}
973+
974+// TestSetCurrentSession_BasicAttachAndSwitch verifies the happy path:
975+// an attached client can set its current session, a second attached
976+// client can target the same session, and one of them can switch to a
977+// different session without disturbing the other's record.
978+func TestSetCurrentSession_BasicAttachAndSwitch(t *testing.T) {
979+ t.Parallel()
980+
981+ b, _ := newTestBackend(t)
982+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-basic")
983+
984+ cidA := newClientID(t)
985+ cidB := newClientID(t)
986+ require.NoError(t, b.AttachClient(ws.ID, cidA))
987+ require.NoError(t, b.AttachClient(ws.ID, cidB))
988+
989+ require.NoError(t, b.SetCurrentSession(ws.ID, cidA, "S1"))
990+ ws.clientsMu.Lock()
991+ require.Equal(t, "S1", ws.clients[cidA].currentSessionID)
992+ ws.clientsMu.Unlock()
993+
994+ require.NoError(t, b.SetCurrentSession(ws.ID, cidB, "S1"))
995+ ws.clientsMu.Lock()
996+ require.Equal(t, "S1", ws.clients[cidA].currentSessionID)
997+ require.Equal(t, "S1", ws.clients[cidB].currentSessionID)
998+ ws.clientsMu.Unlock()
999+
1000+ // B switches to S2; counts redistribute.
1001+ require.NoError(t, b.SetCurrentSession(ws.ID, cidB, "S2"))
1002+ ws.clientsMu.Lock()
1003+ require.Equal(t, "S1", ws.clients[cidA].currentSessionID)
1004+ require.Equal(t, "S2", ws.clients[cidB].currentSessionID)
1005+ ws.clientsMu.Unlock()
1006+
1007+ // A clears its selection.
1008+ require.NoError(t, b.SetCurrentSession(ws.ID, cidA, ""))
1009+ ws.clientsMu.Lock()
1010+ require.Empty(t, ws.clients[cidA].currentSessionID)
1011+ require.Equal(t, "S2", ws.clients[cidB].currentSessionID)
1012+ ws.clientsMu.Unlock()
1013+
1014+ // Drain to release the workspace.
1015+ b.DetachClient(ws.ID, cidA)
1016+ b.DetachClient(ws.ID, cidB)
1017+}
1018+
1019+// TestSetCurrentSession_DetachClearsEntry verifies the implicit
1020+// cleanup: once a client's [clientState] entry is removed (last
1021+// stream closed), its currentSessionID is gone with it.
1022+func TestSetCurrentSession_DetachClearsEntry(t *testing.T) {
1023+ t.Parallel()
1024+
1025+ b, _ := newTestBackend(t)
1026+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-detach")
1027+
1028+ // Anchor client so the workspace is not torn down when cid
1029+ // detaches.
1030+ anchor := newClientID(t)
1031+ require.NoError(t, b.AttachClient(ws.ID, anchor))
1032+
1033+ cid := newClientID(t)
1034+ require.NoError(t, b.AttachClient(ws.ID, cid))
1035+ require.NoError(t, b.SetCurrentSession(ws.ID, cid, "S2"))
1036+
1037+ b.DetachClient(ws.ID, cid)
1038+
1039+ ws.clientsMu.Lock()
1040+ _, present := ws.clients[cid]
1041+ ws.clientsMu.Unlock()
1042+ require.False(t, present, "detach must remove the clientState entry along with its currentSessionID")
1043+
1044+ // A follow-up SetCurrentSession on the gone client must be
1045+ // rejected with ErrClientNotAttached.
1046+ require.ErrorIs(t, b.SetCurrentSession(ws.ID, cid, "S3"), ErrClientNotAttached)
1047+
1048+ b.DetachClient(ws.ID, anchor)
1049+}
1050+
1051+// TestSetCurrentSession_RejectsHoldOnly verifies that a registered
1052+// client whose only claim is a creation hold (streams == 0) cannot
1053+// influence presence: SetCurrentSession returns ErrClientNotAttached
1054+// and the entry's currentSessionID stays empty.
1055+func TestSetCurrentSession_RejectsHoldOnly(t *testing.T) {
1056+ t.Parallel()
1057+
1058+ b, _ := newTestBackend(t)
1059+ // Keep the grace window large so the hold survives the test.
1060+ b.createGrace = time.Hour
1061+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-hold")
1062+
1063+ cid := newClientID(t)
1064+ b.registerClient(ws, cid)
1065+
1066+ require.ErrorIs(t, b.SetCurrentSession(ws.ID, cid, "S1"), ErrClientNotAttached)
1067+
1068+ ws.clientsMu.Lock()
1069+ require.Empty(t, ws.clients[cid].currentSessionID, "hold-only client must not write a session id")
1070+ ws.clientsMu.Unlock()
1071+
1072+ // Drain.
1073+ require.NoError(t, b.releaseHold(ws.ID, cid))
1074+}
1075+
1076+// TestSetCurrentSession_UnknownClient verifies that a client with no
1077+// entry at all is rejected with ErrClientNotAttached.
1078+func TestSetCurrentSession_UnknownClient(t *testing.T) {
1079+ t.Parallel()
1080+
1081+ b, _ := newTestBackend(t)
1082+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-unknown")
1083+
1084+ require.ErrorIs(t, b.SetCurrentSession(ws.ID, newClientID(t), "S1"), ErrClientNotAttached)
1085+}
1086+
1087+// TestSetCurrentSession_RejectsBadInputs covers the validation
1088+// branches: empty/malformed client_id and unknown workspace.
1089+func TestSetCurrentSession_RejectsBadInputs(t *testing.T) {
1090+ t.Parallel()
1091+
1092+ b, _ := newTestBackend(t)
1093+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-bad")
1094+
1095+ require.ErrorIs(t, b.SetCurrentSession(ws.ID, "", "S1"), ErrInvalidClientID)
1096+ require.ErrorIs(t, b.SetCurrentSession(ws.ID, "not-a-uuid", "S1"), ErrInvalidClientID)
1097+
1098+ require.ErrorIs(
1099+ t,
1100+ b.SetCurrentSession("00000000-0000-0000-0000-000000000000", newClientID(t), "S1"),
1101+ ErrWorkspaceNotFound,
1102+ )
1103+}
1104+
1105+// TestSetCurrentSession_RaceWithDetach exercises concurrent
1106+// SetCurrentSession updates from one client racing against detach
1107+// on a second client. The final state must be self-consistent: any
1108+// remaining clientState entries reflect a coherent
1109+// (streams, currentSessionID) pair.
1110+func TestSetCurrentSession_RaceWithDetach(t *testing.T) {
1111+ t.Parallel()
1112+
1113+ b, _ := newTestBackend(t)
1114+ ws, _ := insertTestWorkspace(t, b, "/tmp/current-session-race")
1115+
1116+ cidA := newClientID(t)
1117+ cidB := newClientID(t)
1118+ require.NoError(t, b.AttachClient(ws.ID, cidA))
1119+ require.NoError(t, b.AttachClient(ws.ID, cidB))
1120+
1121+ var wg sync.WaitGroup
1122+ const updates = 200
1123+ wg.Add(3)
1124+ go func() {
1125+ defer wg.Done()
1126+ for i := range updates {
1127+ // Errors are tolerated: once cidA detaches,
1128+ // further updates against cidA must return
1129+ // ErrClientNotAttached but never panic.
1130+ _ = b.SetCurrentSession(ws.ID, cidA, "SA")
1131+ _ = i
1132+ }
1133+ }()
1134+ go func() {
1135+ defer wg.Done()
1136+ for i := range updates {
1137+ _ = b.SetCurrentSession(ws.ID, cidB, "SB")
1138+ _ = i
1139+ }
1140+ }()
1141+ go func() {
1142+ defer wg.Done()
1143+ // Single concurrent detach of cidA partway through.
1144+ b.DetachClient(ws.ID, cidA)
1145+ }()
1146+ wg.Wait()
1147+
1148+ ws.clientsMu.Lock()
1149+ defer ws.clientsMu.Unlock()
1150+ require.NotContains(t, ws.clients, cidA, "detached client must be gone")
1151+ require.Contains(t, ws.clients, cidB, "remaining client must still be present")
1152+ require.Equal(t, "SB", ws.clients[cidB].currentSessionID, "remaining client must keep its last set session")
1153+}
1154+
1155+// TestAttachedClients_BasicLifecycle walks one session's count through
1156+// attach -> set -> second client joins -> switch -> detach. It also
1157+// confirms hold-only and unselected clients do not contribute.
1158+func TestAttachedClients_BasicLifecycle(t *testing.T) {
1159+ t.Parallel()
1160+
1161+ b, _ := newTestBackend(t)
1162+ // Keep the grace window long so the hold-only client survives.
1163+ b.createGrace = time.Hour
1164+ ws, _ := insertTestWorkspace(t, b, "/tmp/attached-clients-basic")
1165+
1166+ // No clients yet.
1167+ n, err := b.AttachedClients(ws.ID, "S1")
1168+ require.NoError(t, err)
1169+ require.Zero(t, n)
1170+
1171+ // Attach A, set to S1. Count for S1 is 1; count for S2 is 0.
1172+ cidA := newClientID(t)
1173+ require.NoError(t, b.AttachClient(ws.ID, cidA))
1174+ require.NoError(t, b.SetCurrentSession(ws.ID, cidA, "S1"))
1175+
1176+ n, err = b.AttachedClients(ws.ID, "S1")
1177+ require.NoError(t, err)
1178+ require.Equal(t, 1, n)
1179+ n, err = b.AttachedClients(ws.ID, "S2")
1180+ require.NoError(t, err)
1181+ require.Zero(t, n)
1182+
1183+ // Attach B, set to S1. Count for S1 is 2.
1184+ cidB := newClientID(t)
1185+ require.NoError(t, b.AttachClient(ws.ID, cidB))
1186+ require.NoError(t, b.SetCurrentSession(ws.ID, cidB, "S1"))
1187+
1188+ n, _ = b.AttachedClients(ws.ID, "S1")
1189+ require.Equal(t, 2, n)
1190+
1191+ // B switches to S2; counts redistribute.
1192+ require.NoError(t, b.SetCurrentSession(ws.ID, cidB, "S2"))
1193+ n, _ = b.AttachedClients(ws.ID, "S1")
1194+ require.Equal(t, 1, n)
1195+ n, _ = b.AttachedClients(ws.ID, "S2")
1196+ require.Equal(t, 1, n)
1197+
1198+ // A hold-only client must NOT be counted, even if we were able to
1199+ // imagine a currentSessionID on it. registerClient leaves
1200+ // currentSessionID empty by construction, and SetCurrentSession
1201+ // rejects hold-only writers — so the contract holds two ways.
1202+ cidHold := newClientID(t)
1203+ b.registerClient(ws, cidHold)
1204+ t.Cleanup(func() { _ = b.releaseHold(ws.ID, cidHold) })
1205+ n, _ = b.AttachedClients(ws.ID, "S1")
1206+ require.Equal(t, 1, n, "hold-only client must not contribute")
1207+ n, _ = b.AttachedClients(ws.ID, "")
1208+ require.Equal(t, 0, n,
1209+ "empty sessionID must not match the hold-only entry (streams==0)")
1210+
1211+ // A client with streams > 0 but currentSessionID == "" is NOT
1212+ // counted toward any non-empty session, and is matched only
1213+ // against the empty session id (which represents the landing
1214+ // screen).
1215+ cidC := newClientID(t)
1216+ require.NoError(t, b.AttachClient(ws.ID, cidC))
1217+ n, _ = b.AttachedClients(ws.ID, "S1")
1218+ require.Equal(t, 1, n, "stream-only client with empty currentSessionID must not be counted toward S1")
1219+ n, _ = b.AttachedClients(ws.ID, "")
1220+ require.Equal(t, 1, n, "stream-only client with empty currentSessionID matches the empty session id")
1221+
1222+ // B detaches: count for S2 drops to 0.
1223+ b.DetachClient(ws.ID, cidB)
1224+ n, _ = b.AttachedClients(ws.ID, "S2")
1225+ require.Zero(t, n)
1226+ n, _ = b.AttachedClients(ws.ID, "S1")
1227+ require.Equal(t, 1, n, "A still on S1")
1228+
1229+ // Final cleanup.
1230+ b.DetachClient(ws.ID, cidA)
1231+ b.DetachClient(ws.ID, cidC)
1232+}
1233+
1234+// TestAttachedClients_UnknownWorkspace verifies the error surface.
1235+func TestAttachedClients_UnknownWorkspace(t *testing.T) {
1236+ t.Parallel()
1237+
1238+ b, _ := newTestBackend(t)
1239+ _, err := b.AttachedClients("00000000-0000-0000-0000-000000000000", "S1")
1240+ require.ErrorIs(t, err, ErrWorkspaceNotFound)
1241+}