git.iko.soy/crush
on main help sign in
fix: race in agent.go (#1853)
f66762bCarlos Alexandro Becker
internal/agent/agent.go+59−48
compare with main
unifiedsplit
@@ -87,12 +87,13 @@ type Model struct {
87}
87}
88
88
89type sessionAgent struct {
89type sessionAgent struct {
90− largeModel Model
90+ largeModel *csync.Value[Model]
91− smallModel Model
91+ smallModel *csync.Value[Model]
92− systemPromptPrefix string
92+ systemPromptPrefix *csync.Value[string]
93− systemPrompt string
93+ systemPrompt *csync.Value[string]
94+ tools *csync.Slice[fantasy.AgentTool]
95+
94 isSubAgent bool
96 isSubAgent bool
95− tools []fantasy.AgentTool
96 sessions session.Service
97 sessions session.Service
97 messages message.Service
98 messages message.Service
98 disableAutoSummarize bool
99 disableAutoSummarize bool
@@ -119,15 +120,15 @@ func NewSessionAgent(
119 opts SessionAgentOptions,
120 opts SessionAgentOptions,
120) SessionAgent {
121) SessionAgent {
121 return &sessionAgent{
122 return &sessionAgent{
122− largeModel: opts.LargeModel,
123+ largeModel: csync.NewValue(opts.LargeModel),
123− smallModel: opts.SmallModel,
124+ smallModel: csync.NewValue(opts.SmallModel),
124− systemPromptPrefix: opts.SystemPromptPrefix,
125+ systemPromptPrefix: csync.NewValue(opts.SystemPromptPrefix),
125− systemPrompt: opts.SystemPrompt,
126+ systemPrompt: csync.NewValue(opts.SystemPrompt),
126 isSubAgent: opts.IsSubAgent,
127 isSubAgent: opts.IsSubAgent,
127 sessions: opts.Sessions,
128 sessions: opts.Sessions,
128 messages: opts.Messages,
129 messages: opts.Messages,
129 disableAutoSummarize: opts.DisableAutoSummarize,
130 disableAutoSummarize: opts.DisableAutoSummarize,
130− tools: opts.Tools,
131+ tools: csync.NewSliceFrom(opts.Tools),
131 isYolo: opts.IsYolo,
132 isYolo: opts.IsYolo,
132 messageQueue: csync.NewMap[string, []SessionAgentCall](),
133 messageQueue: csync.NewMap[string, []SessionAgentCall](),
133 activeRequests: csync.NewMap[string, context.CancelFunc](),
134 activeRequests: csync.NewMap[string, context.CancelFunc](),
@@ -153,15 +154,21 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
153 return nil, nil
154 return nil, nil
154 }
155 }
155
156
157+ // Copy mutable fields under lock to avoid races with SetTools/SetModels.
158+ agentTools := a.tools.Copy()
159+ largeModel := a.largeModel.Get()
160+ systemPrompt := a.systemPrompt.Get()
161+ promptPrefix := a.systemPromptPrefix.Get()
162+
156− if len(a.tools) > 0 {
163+ if len(agentTools) > 0 {
157 // Add Anthropic caching to the last tool.
164 // Add Anthropic caching to the last tool.
158− a.tools[len(a.tools)-1].SetProviderOptions(a.getCacheControlOptions())
165+ agentTools[len(agentTools)-1].SetProviderOptions(a.getCacheControlOptions())
159 }
166 }
160
167
161 agent := fantasy.NewAgent(
168 agent := fantasy.NewAgent(
162− a.largeModel.Model,
169+ largeModel.Model,
163− fantasy.WithSystemPrompt(a.systemPrompt),
170+ fantasy.WithSystemPrompt(systemPrompt),
164− fantasy.WithTools(a.tools...),
171+ fantasy.WithTools(agentTools...),
165 )
172 )
166
173
167 sessionLock := sync.Mutex{}
174 sessionLock := sync.Mutex{}
@@ -234,7 +241,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
234 prepared.Messages = append(prepared.Messages, userMessage.ToAIMessage()...)
241 prepared.Messages = append(prepared.Messages, userMessage.ToAIMessage()...)
235 }
242 }
236
243
237− prepared.Messages = a.workaroundProviderMediaLimitations(prepared.Messages)
244+ prepared.Messages = a.workaroundProviderMediaLimitations(prepared.Messages, largeModel)
238
245
239 lastSystemRoleInx := 0
246 lastSystemRoleInx := 0
240 systemMessageUpdated := false
247 systemMessageUpdated := false
@@ -252,7 +259,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
252 }
259 }
253 }
260 }
254
261
255− if promptPrefix := a.promptPrefix(); promptPrefix != "" {
262+ if promptPrefix != "" {
256 prepared.Messages = append([]fantasy.Message{fantasy.NewSystemMessage(promptPrefix)}, prepared.Messages...)
263 prepared.Messages = append([]fantasy.Message{fantasy.NewSystemMessage(promptPrefix)}, prepared.Messages...)
257 }
264 }
258
265
@@ -260,15 +267,15 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
260 assistantMsg, err = a.messages.Create(callContext, call.SessionID, message.CreateMessageParams{
267 assistantMsg, err = a.messages.Create(callContext, call.SessionID, message.CreateMessageParams{
261 Role: message.Assistant,
268 Role: message.Assistant,
262 Parts: []message.ContentPart{},
269 Parts: []message.ContentPart{},
263− Model: a.largeModel.ModelCfg.Model,
270+ Model: largeModel.ModelCfg.Model,
264− Provider: a.largeModel.ModelCfg.Provider,
271+ Provider: largeModel.ModelCfg.Provider,
265 })
272 })
266 if err != nil {
273 if err != nil {
267 return callContext, prepared, err
274 return callContext, prepared, err
268 }
275 }
269 callContext = context.WithValue(callContext, tools.MessageIDContextKey, assistantMsg.ID)
276 callContext = context.WithValue(callContext, tools.MessageIDContextKey, assistantMsg.ID)
270− callContext = context.WithValue(callContext, tools.SupportsImagesContextKey, a.largeModel.CatwalkCfg.SupportsImages)
277+ callContext = context.WithValue(callContext, tools.SupportsImagesContextKey, largeModel.CatwalkCfg.SupportsImages)
271− callContext = context.WithValue(callContext, tools.ModelNameContextKey, a.largeModel.CatwalkCfg.Name)
278+ callContext = context.WithValue(callContext, tools.ModelNameContextKey, largeModel.CatwalkCfg.Name)
272 currentAssistant = &assistantMsg
279 currentAssistant = &assistantMsg
273 return callContext, prepared, err
280 return callContext, prepared, err
274 },
281 },
@@ -362,7 +369,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
362 sessionLock.Unlock()
369 sessionLock.Unlock()
363 return getSessionErr
370 return getSessionErr
364 }
371 }
365− a.updateSessionUsage(a.largeModel, &updatedSession, stepResult.Usage, a.openrouterCost(stepResult.ProviderMetadata))
372+ a.updateSessionUsage(largeModel, &updatedSession, stepResult.Usage, a.openrouterCost(stepResult.ProviderMetadata))
366 _, sessionErr := a.sessions.Save(genCtx, updatedSession)
373 _, sessionErr := a.sessions.Save(genCtx, updatedSession)
367 sessionLock.Unlock()
374 sessionLock.Unlock()
368 if sessionErr != nil {
375 if sessionErr != nil {
@@ -372,7 +379,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
372 },
379 },
373 StopWhen: []fantasy.StopCondition{
380 StopWhen: []fantasy.StopCondition{
374 func(_ []fantasy.StepResult) bool {
381 func(_ []fantasy.StepResult) bool {
375− cw := int64(a.largeModel.CatwalkCfg.ContextWindow)
382+ cw := int64(largeModel.CatwalkCfg.ContextWindow)
376 tokens := currentSession.CompletionTokens + currentSession.PromptTokens
383 tokens := currentSession.CompletionTokens + currentSession.PromptTokens
377 remaining := cw - tokens
384 remaining := cw - tokens
378 var threshold int64
385 var threshold int64
@@ -474,7 +481,7 @@ func (a *sessionAgent) Run(ctx context.Context, call SessionAgentCall) (*fantasy
474 currentAssistant.AddFinish(
481 currentAssistant.AddFinish(
475 message.FinishReasonError,
482 message.FinishReasonError,
476 "Copilot model not enabled",
483 "Copilot model not enabled",
477− fmt.Sprintf("%q is not enabled in Copilot. Go to the following page to enable it. Then, wait 5 minutes before trying again. %s", a.largeModel.CatwalkCfg.Name, link),
484+ fmt.Sprintf("%q is not enabled in Copilot. Go to the following page to enable it. Then, wait 5 minutes before trying again. %s", largeModel.CatwalkCfg.Name, link),
478 )
485 )
479 } else {
486 } else {
480 currentAssistant.AddFinish(message.FinishReasonError, cmp.Or(stringext.Capitalize(providerErr.Title), defaultTitle), providerErr.Message)
487 currentAssistant.AddFinish(message.FinishReasonError, cmp.Or(stringext.Capitalize(providerErr.Title), defaultTitle), providerErr.Message)
@@ -529,6 +536,10 @@ func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fan
529 return ErrSessionBusy
536 return ErrSessionBusy
530 }
537 }
531
538
539+ // Copy mutable fields under lock to avoid races with SetModels.
540+ largeModel := a.largeModel.Get()
541+ systemPromptPrefix := a.systemPromptPrefix.Get()
542+
532 currentSession, err := a.sessions.Get(ctx, sessionID)
543 currentSession, err := a.sessions.Get(ctx, sessionID)
533 if err != nil {
544 if err != nil {
534 return fmt.Errorf("failed to get session: %w", err)
545 return fmt.Errorf("failed to get session: %w", err)
@@ -549,13 +560,13 @@ func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fan
549 defer a.activeRequests.Del(sessionID)
560 defer a.activeRequests.Del(sessionID)
550 defer cancel()
561 defer cancel()
551
562
552− agent := fantasy.NewAgent(a.largeModel.Model,
563+ agent := fantasy.NewAgent(largeModel.Model,
553 fantasy.WithSystemPrompt(string(summaryPrompt)),
564 fantasy.WithSystemPrompt(string(summaryPrompt)),
554 )
565 )
555 summaryMessage, err := a.messages.Create(ctx, sessionID, message.CreateMessageParams{
566 summaryMessage, err := a.messages.Create(ctx, sessionID, message.CreateMessageParams{
556 Role: message.Assistant,
567 Role: message.Assistant,
557− Model: a.largeModel.Model.Model(),
568+ Model: largeModel.Model.Model(),
558− Provider: a.largeModel.Model.Provider(),
569+ Provider: largeModel.Model.Provider(),
559 IsSummaryMessage: true,
570 IsSummaryMessage: true,
560 })
571 })
561 if err != nil {
572 if err != nil {
@@ -570,8 +581,8 @@ func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fan
570 ProviderOptions: opts,
581 ProviderOptions: opts,
571 PrepareStep: func(callContext context.Context, options fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
582 PrepareStep: func(callContext context.Context, options fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
572 prepared.Messages = options.Messages
583 prepared.Messages = options.Messages
573− if a.systemPromptPrefix != "" {
584+ if systemPromptPrefix != "" {
574− prepared.Messages = append([]fantasy.Message{fantasy.NewSystemMessage(a.systemPromptPrefix)}, prepared.Messages...)
585+ prepared.Messages = append([]fantasy.Message{fantasy.NewSystemMessage(systemPromptPrefix)}, prepared.Messages...)
575 }
586 }
576 return callContext, prepared, nil
587 return callContext, prepared, nil
577 },
588 },
@@ -622,7 +633,7 @@ func (a *sessionAgent) Summarize(ctx context.Context, sessionID string, opts fan
622 }
633 }
623 }
634 }
624
635
625− a.updateSessionUsage(a.largeModel, &currentSession, resp.TotalUsage, openrouterCost)
636+ a.updateSessionUsage(largeModel, &currentSession, resp.TotalUsage, openrouterCost)
626
637
627 // Just in case, get just the last usage info.
638 // Just in case, get just the last usage info.
628 usage := resp.Response.Usage
639 usage := resp.Response.Usage
@@ -730,9 +741,13 @@ func (a *sessionAgent) generateTitle(ctx context.Context, sessionID string, user
730 return
741 return
731 }
742 }
732
743
744+ smallModel := a.smallModel.Get()
745+ largeModel := a.largeModel.Get()
746+ systemPromptPrefix := a.systemPromptPrefix.Get()
747+
733 var maxOutputTokens int64 = 40
748 var maxOutputTokens int64 = 40
734− if a.smallModel.CatwalkCfg.CanReason {
749+ if smallModel.CatwalkCfg.CanReason {
735− maxOutputTokens = a.smallModel.CatwalkCfg.DefaultMaxTokens
750+ maxOutputTokens = smallModel.CatwalkCfg.DefaultMaxTokens
736 }
751 }
737
752
738 newAgent := func(m fantasy.LanguageModel, p []byte, tok int64) fantasy.Agent {
753 newAgent := func(m fantasy.LanguageModel, p []byte, tok int64) fantasy.Agent {
@@ -746,9 +761,9 @@ func (a *sessionAgent) generateTitle(ctx context.Context, sessionID string, user
746 Prompt: fmt.Sprintf("Generate a concise title for the following content:\n\n%s\n <think>\n\n</think>", userPrompt),
761 Prompt: fmt.Sprintf("Generate a concise title for the following content:\n\n%s\n <think>\n\n</think>", userPrompt),
747 PrepareStep: func(callCtx context.Context, opts fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
762 PrepareStep: func(callCtx context.Context, opts fantasy.PrepareStepFunctionOptions) (_ context.Context, prepared fantasy.PrepareStepResult, err error) {
748 prepared.Messages = opts.Messages
763 prepared.Messages = opts.Messages
749− if a.systemPromptPrefix != "" {
764+ if systemPromptPrefix != "" {
750 prepared.Messages = append([]fantasy.Message{
765 prepared.Messages = append([]fantasy.Message{
751− fantasy.NewSystemMessage(a.systemPromptPrefix),
766+ fantasy.NewSystemMessage(systemPromptPrefix),
752 }, prepared.Messages...)
767 }, prepared.Messages...)
753 }
768 }
754 return callCtx, prepared, nil
769 return callCtx, prepared, nil
@@ -756,7 +771,7 @@ func (a *sessionAgent) generateTitle(ctx context.Context, sessionID string, user
756 }
771 }
757
772
758 // Use the small model to generate the title.
773 // Use the small model to generate the title.
759− model := &a.smallModel
774+ model := smallModel
760 agent := newAgent(model.Model, titlePrompt, maxOutputTokens)
775 agent := newAgent(model.Model, titlePrompt, maxOutputTokens)
761 resp, err := agent.Stream(ctx, streamCall)
776 resp, err := agent.Stream(ctx, streamCall)
762 if err == nil {
777 if err == nil {
@@ -765,7 +780,7 @@ func (a *sessionAgent) generateTitle(ctx context.Context, sessionID string, user
765 } else {
780 } else {
766 // It didn't work. Let's try with the big model.
781 // It didn't work. Let's try with the big model.
767 slog.Error("error generating title with small model; trying big model", "err", err)
782 slog.Error("error generating title with small model; trying big model", "err", err)
768− model = &a.largeModel
783+ model = largeModel
769 agent = newAgent(model.Model, titlePrompt, maxOutputTokens)
784 agent = newAgent(model.Model, titlePrompt, maxOutputTokens)
770 resp, err = agent.Stream(ctx, streamCall)
785 resp, err = agent.Stream(ctx, streamCall)
771 if err == nil {
786 if err == nil {
@@ -960,24 +975,20 @@ func (a *sessionAgent) QueuedPromptsList(sessionID string) []string {
960}
975}
961
976
962func (a *sessionAgent) SetModels(large Model, small Model) {
977func (a *sessionAgent) SetModels(large Model, small Model) {
963− a.largeModel = large
978+ a.largeModel.Set(large)
964− a.smallModel = small
979+ a.smallModel.Set(small)
965}
980}
966
981
967func (a *sessionAgent) SetTools(tools []fantasy.AgentTool) {
982func (a *sessionAgent) SetTools(tools []fantasy.AgentTool) {
968− a.tools = tools
983+ a.tools.SetSlice(tools)
969}
984}
970
985
971func (a *sessionAgent) SetSystemPrompt(systemPrompt string) {
986func (a *sessionAgent) SetSystemPrompt(systemPrompt string) {
972− a.systemPrompt = systemPrompt
987+ a.systemPrompt.Set(systemPrompt)
973}
988}
974
989
975func (a *sessionAgent) Model() Model {
990func (a *sessionAgent) Model() Model {
976− return a.largeModel
991+ return a.largeModel.Get()
977−}
978−
979−func (a *sessionAgent) promptPrefix() string {
980− return a.systemPromptPrefix
981}
992}
982
993
983// convertToToolResult converts a fantasy tool result to a message tool result.
994// convertToToolResult converts a fantasy tool result to a message tool result.
@@ -1034,9 +1045,9 @@ func (a *sessionAgent) convertToToolResult(result fantasy.ToolResultContent) mes
1034//
1045//
1035// BEFORE: [tool result: image data]
1046// BEFORE: [tool result: image data]
1036// AFTER: [tool result: "Image loaded - see attached"], [user: image attachment]
1047// AFTER: [tool result: "Image loaded - see attached"], [user: image attachment]
1037−func (a *sessionAgent) workaroundProviderMediaLimitations(messages []fantasy.Message) []fantasy.Message {
1048+func (a *sessionAgent) workaroundProviderMediaLimitations(messages []fantasy.Message, largeModel Model) []fantasy.Message {
1038− providerSupportsMedia := a.largeModel.ModelCfg.Provider == string(catwalk.InferenceProviderAnthropic) ||
1049+ providerSupportsMedia := largeModel.ModelCfg.Provider == string(catwalk.InferenceProviderAnthropic) ||
1039− a.largeModel.ModelCfg.Provider == string(catwalk.InferenceProviderBedrock)
1050+ largeModel.ModelCfg.Provider == string(catwalk.InferenceProviderBedrock)
1040
1051
1041 if providerSupportsMedia {
1052 if providerSupportsMedia {
1042 return messages
1053 return messages
internal/agent/event.go+6−6
@@ -7,23 +7,23 @@ import (
7 "github.com/charmbracelet/crush/internal/event"
7 "github.com/charmbracelet/crush/internal/event"
8)
8)
9
9
10−func (a sessionAgent) eventPromptSent(sessionID string) {
10+func (a *sessionAgent) eventPromptSent(sessionID string) {
11 event.PromptSent(
11 event.PromptSent(
12− a.eventCommon(sessionID, a.largeModel)...,
12+ a.eventCommon(sessionID, a.largeModel.Get())...,
13 )
13 )
14}
14}
15
15
16−func (a sessionAgent) eventPromptResponded(sessionID string, duration time.Duration) {
16+func (a *sessionAgent) eventPromptResponded(sessionID string, duration time.Duration) {
17 event.PromptResponded(
17 event.PromptResponded(
18 append(
18 append(
19− a.eventCommon(sessionID, a.largeModel),
19+ a.eventCommon(sessionID, a.largeModel.Get()),
20 "prompt duration pretty", duration.String(),
20 "prompt duration pretty", duration.String(),
21 "prompt duration in seconds", int64(duration.Seconds()),
21 "prompt duration in seconds", int64(duration.Seconds()),
22 )...,
22 )...,
23 )
23 )
24}
24}
25
25
26−func (a sessionAgent) eventTokensUsed(sessionID string, model Model, usage fantasy.Usage, cost float64) {
26+func (a *sessionAgent) eventTokensUsed(sessionID string, model Model, usage fantasy.Usage, cost float64) {
27 event.TokensUsed(
27 event.TokensUsed(
28 append(
28 append(
29 a.eventCommon(sessionID, model),
29 a.eventCommon(sessionID, model),
@@ -37,7 +37,7 @@ func (a sessionAgent) eventTokensUsed(sessionID string, model Model, usage fanta
37 )
37 )
38}
38}
39
39
40−func (a sessionAgent) eventCommon(sessionID string, model Model) []any {
40+func (a *sessionAgent) eventCommon(sessionID string, model Model) []any {
41 m := model.ModelCfg
41 m := model.ModelCfg
42
42
43 return []any{
43 return []any{
internal/csync/slices.go+10−34
@@ -2,7 +2,6 @@ package csync
2
2
3import (
3import (
4 "iter"
4 "iter"
5− "slices"
6 "sync"
5 "sync"
7)
6)
8
7
@@ -63,24 +62,6 @@ func (s *Slice[T]) Append(items ...T) {
63 s.inner = append(s.inner, items...)
62 s.inner = append(s.inner, items...)
64}
63}
65
64
66−// Prepend adds an element to the beginning of the slice.
67−func (s *Slice[T]) Prepend(item T) {
68− s.mu.Lock()
69− defer s.mu.Unlock()
70− s.inner = append([]T{item}, s.inner...)
71−}
72−
73−// Delete removes the element at the specified index.
74−func (s *Slice[T]) Delete(index int) bool {
75− s.mu.Lock()
76− defer s.mu.Unlock()
77− if index < 0 || index >= len(s.inner) {
78− return false
79− }
80− s.inner = slices.Delete(s.inner, index, index+1)
81− return true
82−}
83−
84// Get returns the element at the specified index.
65// Get returns the element at the specified index.
85func (s *Slice[T]) Get(index int) (T, bool) {
66func (s *Slice[T]) Get(index int) (T, bool) {
86 s.mu.RLock()
67 s.mu.RLock()
@@ -92,17 +73,6 @@ func (s *Slice[T]) Get(index int) (T, bool) {
92 return s.inner[index], true
73 return s.inner[index], true
93}
74}
94
75
95−// Set updates the element at the specified index.
96−func (s *Slice[T]) Set(index int, item T) bool {
97− s.mu.Lock()
98− defer s.mu.Unlock()
99− if index < 0 || index >= len(s.inner) {
100− return false
101− }
102− s.inner[index] = item
103− return true
104−}
105−
106// Len returns the number of elements in the slice.
76// Len returns the number of elements in the slice.
107func (s *Slice[T]) Len() int {
77func (s *Slice[T]) Len() int {
108 s.mu.RLock()
78 s.mu.RLock()
@@ -131,10 +101,7 @@ func (s *Slice[T]) Seq() iter.Seq[T] {
131
101
132// Seq2 returns an iterator that yields index-value pairs from the slice.
102// Seq2 returns an iterator that yields index-value pairs from the slice.
133func (s *Slice[T]) Seq2() iter.Seq2[int, T] {
103func (s *Slice[T]) Seq2() iter.Seq2[int, T] {
134− s.mu.RLock()
135− items := make([]T, len(s.inner))
104+ items := s.Copy()
136− copy(items, s.inner)
137− s.mu.RUnlock()
138 return func(yield func(int, T) bool) {
105 return func(yield func(int, T) bool) {
139 for i, v := range items {
106 for i, v := range items {
140 if !yield(i, v) {
107 if !yield(i, v) {
@@ -143,3 +110,12 @@ func (s *Slice[T]) Seq2() iter.Seq2[int, T] {
143 }
110 }
144 }
111 }
145}
112}
113+
114+// Copy returns a copy of the inner slice.
115+func (s *Slice[T]) Copy() []T {
116+ s.mu.RLock()
117+ defer s.mu.RUnlock()
118+ items := make([]T, len(s.inner))
119+ copy(items, s.inner)
120+ return items
121+}
internal/csync/slices_test.go−57
@@ -109,44 +109,6 @@ func TestSlice(t *testing.T) {
109 require.Equal(t, "world", val)
109 require.Equal(t, "world", val)
110 })
110 })
111
111
112− t.Run("Prepend", func(t *testing.T) {
113− s := NewSlice[string]()
114− s.Append("world")
115− s.Prepend("hello")
116−
117− require.Equal(t, 2, s.Len())
118− val, ok := s.Get(0)
119− require.True(t, ok)
120− require.Equal(t, "hello", val)
121−
122− val, ok = s.Get(1)
123− require.True(t, ok)
124− require.Equal(t, "world", val)
125− })
126−
127− t.Run("Delete", func(t *testing.T) {
128− s := NewSliceFrom([]int{1, 2, 3, 4, 5})
129−
130− // Delete middle element
131− ok := s.Delete(2)
132− require.True(t, ok)
133− require.Equal(t, 4, s.Len())
134−
135− expected := []int{1, 2, 4, 5}
136− actual := slices.Collect(s.Seq())
137− require.Equal(t, expected, actual)
138−
139− // Delete out of bounds
140− ok = s.Delete(10)
141− require.False(t, ok)
142− require.Equal(t, 4, s.Len())
143−
144− // Delete negative index
145− ok = s.Delete(-1)
146− require.False(t, ok)
147− require.Equal(t, 4, s.Len())
148− })
149−
150 t.Run("Get", func(t *testing.T) {
112 t.Run("Get", func(t *testing.T) {
151 s := NewSliceFrom([]string{"a", "b", "c"})
113 s := NewSliceFrom([]string{"a", "b", "c"})
152
114
@@ -163,25 +125,6 @@ func TestSlice(t *testing.T) {
163 require.False(t, ok)
125 require.False(t, ok)
164 })
126 })
165
127
166− t.Run("Set", func(t *testing.T) {
167− s := NewSliceFrom([]string{"a", "b", "c"})
168−
169− ok := s.Set(1, "modified")
170− require.True(t, ok)
171−
172− val, ok := s.Get(1)
173− require.True(t, ok)
174− require.Equal(t, "modified", val)
175−
176− // Out of bounds
177− ok = s.Set(10, "invalid")
178− require.False(t, ok)
179−
180− // Negative index
181− ok = s.Set(-1, "invalid")
182− require.False(t, ok)
183− })
184−
185 t.Run("SetSlice", func(t *testing.T) {
128 t.Run("SetSlice", func(t *testing.T) {
186 s := NewSlice[int]()
129 s := NewSlice[int]()
187 s.Append(1)
130 s.Append(1)
internal/csync/value.gonew file+44
@@ -0,0 +1,44 @@
1+package csync
2+
3+import (
4+ "reflect"
5+ "sync"
6+)
7+
8+// Value is a generic thread-safe wrapper for any value type.
9+//
10+// For slices, use [Slice]. For maps, use [Map]. Pointers are not supported.
11+type Value[T any] struct {
12+ v T
13+ mu sync.RWMutex
14+}
15+
16+// NewValue creates a new Value with the given initial value.
17+//
18+// Panics if t is a pointer, slice, or map. Use the dedicated types for those.
19+func NewValue[T any](t T) *Value[T] {
20+ v := reflect.ValueOf(t)
21+ switch v.Kind() {
22+ case reflect.Pointer:
23+ panic("csync.Value does not support pointer types")
24+ case reflect.Slice:
25+ panic("csync.Value does not support slice types; use csync.Slice")
26+ case reflect.Map:
27+ panic("csync.Value does not support map types; use csync.Map")
28+ }
29+ return &Value[T]{v: t}
30+}
31+
32+// Get returns the current value.
33+func (v *Value[T]) Get() T {
34+ v.mu.RLock()
35+ defer v.mu.RUnlock()
36+ return v.v
37+}
38+
39+// Set updates the value.
40+func (v *Value[T]) Set(t T) {
41+ v.mu.Lock()
42+ defer v.mu.Unlock()
43+ v.v = t
44+}
internal/csync/value_test.gonew file+99
@@ -0,0 +1,99 @@
1+package csync
2+
3+import (
4+ "sync"
5+ "testing"
6+
7+ "github.com/stretchr/testify/require"
8+)
9+
10+func TestValue_GetSet(t *testing.T) {
11+ t.Parallel()
12+
13+ v := NewValue(42)
14+ require.Equal(t, 42, v.Get())
15+
16+ v.Set(100)
17+ require.Equal(t, 100, v.Get())
18+}
19+
20+func TestValue_ZeroValue(t *testing.T) {
21+ t.Parallel()
22+
23+ v := NewValue("")
24+ require.Equal(t, "", v.Get())
25+
26+ v.Set("hello")
27+ require.Equal(t, "hello", v.Get())
28+}
29+
30+func TestValue_Struct(t *testing.T) {
31+ t.Parallel()
32+
33+ type config struct {
34+ Name string
35+ Count int
36+ }
37+
38+ v := NewValue(config{Name: "test", Count: 1})
39+ require.Equal(t, config{Name: "test", Count: 1}, v.Get())
40+
41+ v.Set(config{Name: "updated", Count: 2})
42+ require.Equal(t, config{Name: "updated", Count: 2}, v.Get())
43+}
44+
45+func TestValue_PointerPanics(t *testing.T) {
46+ t.Parallel()
47+
48+ require.Panics(t, func() {
49+ NewValue(&struct{}{})
50+ })
51+}
52+
53+func TestValue_SlicePanics(t *testing.T) {
54+ t.Parallel()
55+
56+ require.Panics(t, func() {
57+ NewValue([]string{"a", "b"})
58+ })
59+}
60+
61+func TestValue_MapPanics(t *testing.T) {
62+ t.Parallel()
63+
64+ require.Panics(t, func() {
65+ NewValue(map[string]int{"a": 1})
66+ })
67+}
68+
69+func TestValue_ConcurrentAccess(t *testing.T) {
70+ t.Parallel()
71+
72+ v := NewValue(0)
73+ var wg sync.WaitGroup
74+
75+ // Concurrent writers.
76+ for i := range 100 {
77+ wg.Add(1)
78+ go func(val int) {
79+ defer wg.Done()
80+ v.Set(val)
81+ }(i)
82+ }
83+
84+ // Concurrent readers.
85+ for range 100 {
86+ wg.Add(1)
87+ go func() {
88+ defer wg.Done()
89+ _ = v.Get()
90+ }()
91+ }
92+
93+ wg.Wait()
94+
95+ // Value should be one of the set values (0-99).
96+ got := v.Get()
97+ require.GreaterOrEqual(t, got, 0)
98+ require.Less(t, got, 100)
99+}