refactor: replace test-overridden globals with parallel-safe injection

cmd/root/flags.go

@@ -26,7 +26,7 @@ const (
 )
 
 func addRuntimeConfigFlags(cmd *cobra.Command, runConfig *config.RuntimeConfig) {
-	addGatewayFlags(cmd, runConfig)
+	addGatewayFlags(cmd, runConfig, userconfig.Load)
 	cmd.PersistentFlags().StringSliceVar(&runConfig.EnvFiles, "env-from-file", nil, "Set environment variables from file")
 	cmd.PersistentFlags().BoolVar(&runConfig.GlobalCodeMode, "code-mode-tools", false, "Provide a single tool to call other tools via Javascript")
 	cmd.PersistentFlags().StringVar(&runConfig.WorkingDir, "working-dir", "", "Set the working directory for the session (applies to tools and relative paths)")
@@ -73,11 +73,12 @@ func canonize(endpoint string) string {
 	return strings.TrimSuffix(strings.TrimSpace(endpoint), "/")
 }
 
-// loadUserConfig is the function used to load user configuration.
-// It can be overridden in tests.
-var loadUserConfig = userconfig.Load
+// userConfigLoader loads the user configuration. Production passes
+// [userconfig.Load]; tests inject a stub so they neither touch the real
+// config file nor share a mutable package global (and can run in parallel).
+type userConfigLoader func() (*userconfig.Config, error)
 
-func addGatewayFlags(cmd *cobra.Command, runConfig *config.RuntimeConfig) {
+func addGatewayFlags(cmd *cobra.Command, runConfig *config.RuntimeConfig, loadUserConfig userConfigLoader) {
 	cmd.PersistentFlags().StringVar(&runConfig.ModelsGateway, flagModelsGateway, "", "Set the models gateway address")
 
 	persistentPreRunE := cmd.PersistentPreRunE

cmd/root/flags_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/docker/docker-agent/pkg/config"
 	"github.com/docker/docker-agent/pkg/config/latest"
+	"github.com/docker/docker-agent/pkg/environment"
 	"github.com/docker/docker-agent/pkg/userconfig"
 )
 
@@ -70,26 +71,30 @@ func TestGatewayLogic(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			t.Setenv("CAGENT_MODELS_GATEWAY", tt.env) // Make sure the user doesn't have an old env set
-			t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", tt.env)
+			t.Parallel()
 
-			// Mock user config loader
-			original := loadUserConfig
-			loadUserConfig = func() (*userconfig.Config, error) {
+			// Inject env via a map provider rather than t.Setenv so the
+			// test stays parallel-safe.
+			loadUserConfig := func() (*userconfig.Config, error) {
 				if tt.userConfig != nil {
 					return tt.userConfig, nil
 				}
 				return &userconfig.Config{}, nil
 			}
-			t.Cleanup(func() { loadUserConfig = original })
 
 			cmd := &cobra.Command{
 				RunE: func(*cobra.Command, []string) error {
 					return nil
 				},
 			}
-			runConfig := config.RuntimeConfig{}
-			addGatewayFlags(cmd, &runConfig)
+			env := map[string]string{}
+			if tt.env != "" {
+				env[envModelsGateway] = tt.env
+			}
+			runConfig := config.RuntimeConfig{
+				EnvProviderForTests: environment.NewMapEnvProvider(env),
+			}
+			addGatewayFlags(cmd, &runConfig, loadUserConfig)
 
 			cmd.SetArgs(tt.args)
 			err := cmd.Execute()
@@ -123,7 +128,7 @@ func TestGatewayFlags_CallsAncestorPersistentPreRunE(t *testing.T) {
 		RunE: func(*cobra.Command, []string) error { return nil },
 	}
 	runConfig := config.RuntimeConfig{}
-	addGatewayFlags(leaf, &runConfig)
+	addGatewayFlags(leaf, &runConfig, userconfig.Load)
 
 	middle.AddCommand(leaf)
 	root.AddCommand(middle)
@@ -166,7 +171,7 @@ func TestGatewayFlags_RunsParentBeforeMaterialisingEnvProvider(t *testing.T) {
 	runConfig := config.RuntimeConfig{
 		EnvProviderForTests: &recordingProvider{onGet: func() { envProviderConsulted = true }},
 	}
-	addGatewayFlags(leaf, &runConfig)
+	addGatewayFlags(leaf, &runConfig, userconfig.Load)
 	root.AddCommand(leaf)
 
 	root.SetArgs([]string{"leaf"})
@@ -285,25 +290,28 @@ func TestDefaultModelLogic(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", tt.env)
+			t.Parallel()
 
-			// Mock user config loader
-			original := loadUserConfig
-			loadUserConfig = func() (*userconfig.Config, error) {
+			loadUserConfig := func() (*userconfig.Config, error) {
 				if tt.userConfig != nil {
 					return tt.userConfig, nil
 				}
 				return &userconfig.Config{}, nil
 			}
-			t.Cleanup(func() { loadUserConfig = original })
 
 			cmd := &cobra.Command{
 				RunE: func(*cobra.Command, []string) error {
 					return nil
 				},
 			}
-			runConfig := config.RuntimeConfig{}
-			addGatewayFlags(cmd, &runConfig)
+			env := map[string]string{}
+			if tt.env != "" {
+				env[envDefaultModel] = tt.env
+			}
+			runConfig := config.RuntimeConfig{
+				EnvProviderForTests: environment.NewMapEnvProvider(env),
+			}
+			addGatewayFlags(cmd, &runConfig, loadUserConfig)
 
 			cmd.SetArgs(nil)
 			err := cmd.Execute()

cmd/root/models.go

@@ -21,6 +21,7 @@ import (
 	"github.com/docker/docker-agent/pkg/httpclient"
 	"github.com/docker/docker-agent/pkg/model/provider"
 	"github.com/docker/docker-agent/pkg/telemetry"
+	"github.com/docker/docker-agent/pkg/userconfig"
 )
 
 type modelsListFlags struct {
@@ -96,7 +97,7 @@ func newModelsListCmd() *cobra.Command {
 	cmd.Flags().StringVarP(&flags.providerFilter, "provider", "p", "", "Filter by provider name")
 	cmd.Flags().StringVar(&flags.format, "format", "table", "Output format: table, json")
 	cmd.Flags().BoolVarP(&flags.all, "all", "a", false, "Include models from all providers, not just those with credentials")
-	addGatewayFlags(cmd, &flags.runConfig)
+	addGatewayFlags(cmd, &flags.runConfig, userconfig.Load)
 
 	return cmd
 }

cmd/root/models_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/docker/docker-agent/pkg/config"
-	"github.com/docker/docker-agent/pkg/userconfig"
+	"github.com/docker/docker-agent/pkg/paths"
 )
 
 func TestModelsListCommand_DefaultOutput(t *testing.T) {
@@ -24,9 +24,10 @@ func TestModelsListCommand_DefaultOutput(t *testing.T) {
 	t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", "")
 	t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", "")
 
-	original := loadUserConfig
-	loadUserConfig = func() (*userconfig.Config, error) { return &userconfig.Config{}, nil }
-	t.Cleanup(func() { loadUserConfig = original })
+	// Point at an empty config dir so userconfig.Load returns an empty
+	// config without reading the developer's real ~/.config/cagent.
+	paths.SetConfigDir(t.TempDir())
+	t.Cleanup(func() { paths.SetConfigDir("") })
 
 	var buf bytes.Buffer
 	cmd := newModelsCmd()
@@ -49,9 +50,10 @@ func TestModelsListCommand_ProviderFilter(t *testing.T) {
 	t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", "")
 	t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", "")
 
-	original := loadUserConfig
-	loadUserConfig = func() (*userconfig.Config, error) { return &userconfig.Config{}, nil }
-	t.Cleanup(func() { loadUserConfig = original })
+	// Point at an empty config dir so userconfig.Load returns an empty
+	// config without reading the developer's real ~/.config/cagent.
+	paths.SetConfigDir(t.TempDir())
+	t.Cleanup(func() { paths.SetConfigDir("") })
 
 	var buf bytes.Buffer
 	cmd := newModelsCmd()
@@ -79,9 +81,10 @@ func TestModelsListCommand_JSONFormat(t *testing.T) {
 	t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", "")
 	t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", "")
 
-	original := loadUserConfig
-	loadUserConfig = func() (*userconfig.Config, error) { return &userconfig.Config{}, nil }
-	t.Cleanup(func() { loadUserConfig = original })
+	// Point at an empty config dir so userconfig.Load returns an empty
+	// config without reading the developer's real ~/.config/cagent.
+	paths.SetConfigDir(t.TempDir())
+	t.Cleanup(func() { paths.SetConfigDir("") })
 
 	var buf bytes.Buffer
 	cmd := newModelsCmd()
@@ -114,9 +117,10 @@ func TestModelsListCommand_DefaultMarker(t *testing.T) {
 	t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", "")
 	t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", "")
 
-	original := loadUserConfig
-	loadUserConfig = func() (*userconfig.Config, error) { return &userconfig.Config{}, nil }
-	t.Cleanup(func() { loadUserConfig = original })
+	// Point at an empty config dir so userconfig.Load returns an empty
+	// config without reading the developer's real ~/.config/cagent.
+	paths.SetConfigDir(t.TempDir())
+	t.Cleanup(func() { paths.SetConfigDir("") })
 
 	var buf bytes.Buffer
 	cmd := newModelsCmd()
@@ -307,9 +311,10 @@ func TestModelsListCommand_NoCredentials(t *testing.T) {
 	t.Setenv("DOCKER_AGENT_MODELS_GATEWAY", "")
 	t.Setenv("DOCKER_AGENT_DEFAULT_MODEL", "")
 
-	original := loadUserConfig
-	loadUserConfig = func() (*userconfig.Config, error) { return &userconfig.Config{}, nil }
-	t.Cleanup(func() { loadUserConfig = original })
+	// Point at an empty config dir so userconfig.Load returns an empty
+	// config without reading the developer's real ~/.config/cagent.
+	paths.SetConfigDir(t.TempDir())
+	t.Cleanup(func() { paths.SetConfigDir("") })
 
 	var buf bytes.Buffer
 	cmd := newModelsCmd()

pkg/runtime/fallback.go

@@ -333,7 +333,7 @@ func (e *fallbackExecutor) execute(
 				}
 			}
 
-			res, err := handleStream(streamCtx, streamCancel, stream, a, agentTools, sess, m, e.telemetry, events)
+			res, err := handleStream(streamCtx, streamCancel, stream, a, agentTools, sess, m, e.telemetry, events, defaultStreamIdleTimeout)
 			streamCancel(nil) // always release the child context
 			if err != nil {
 				lastErr = err

pkg/runtime/streaming.go

@@ -26,9 +26,10 @@ import (
 // for tens of seconds during extended thinking. The timeout only fires when
 // absolutely no bytes arrive at the SSE layer (not just no new tokens).
 //
-// This is a var rather than a const so tests can shorten it without changing
-// production behaviour.
-var defaultStreamIdleTimeout = 5 * time.Minute
+// handleStream takes the idle timeout as a parameter rather than reading this
+// constant directly, so tests can pass a short timeout without mutating shared
+// state (and can therefore run in parallel).
+const defaultStreamIdleTimeout = 5 * time.Minute
 
 // errStreamIdle is the sentinel error returned when the upstream model stream
 // produces no SSE events for longer than defaultStreamIdleTimeout. It does
@@ -61,12 +62,15 @@ type streamResult struct {
 // passed to CreateChatCompletionStream so that Go's HTTP transport closes the
 // underlying TCP connection and unblocks the stream reader goroutine.
 //
+// idleTimeout bounds the wait for the next SSE chunk; production callers pass
+// defaultStreamIdleTimeout, tests pass a short value.
+//
 // handleStream is a pure stream-aggregation routine: it does not touch
 // runtime state and can be unit-tested by feeding a mock chat.MessageStream.
 // It is intentionally a free function rather than a method on *LocalRuntime
 // so the dependency direction is explicit (the loop calls into the chunker,
 // never the reverse).
-func handleStream(ctx context.Context, cancelStream context.CancelCauseFunc, stream chat.MessageStream, a *agent.Agent, agentTools []tools.Tool, sess *session.Session, m *modelsdev.Model, tel Telemetry, events EventSink) (streamResult, error) {
+func handleStream(ctx context.Context, cancelStream context.CancelCauseFunc, stream chat.MessageStream, a *agent.Agent, agentTools []tools.Tool, sess *session.Session, m *modelsdev.Model, tel Telemetry, events EventSink, idleTimeout time.Duration) (streamResult, error) {
 	// done is closed when handleStream exits (for any reason) so the reader
 	// goroutine below can detect it and stop trying to send on recvCh.
 	done := make(chan struct{})
@@ -98,7 +102,7 @@ func handleStream(ctx context.Context, cancelStream context.CancelCauseFunc, str
 		}
 	}()
 
-	idleTimer := time.NewTimer(defaultStreamIdleTimeout)
+	idleTimer := time.NewTimer(idleTimeout)
 	defer idleTimer.Stop()
 
 	var fullContent strings.Builder
@@ -174,7 +178,7 @@ mainLoop:
 				default:
 				}
 			}
-			idleTimer.Reset(defaultStreamIdleTimeout)
+			idleTimer.Reset(idleTimeout)
 
 			if errors.Is(res.err, io.EOF) {
 				break mainLoop
@@ -339,15 +343,15 @@ mainLoop:
 			slog.WarnContext(ctx, "Model stream stalled: no data received within idle timeout",
 				"agent", a.Name(),
 				"session_id", sess.ID,
-				"idle_timeout", defaultStreamIdleTimeout,
+				"idle_timeout", idleTimeout,
 			)
 			// Cancel the HTTP request context so Go's transport closes the
 			// underlying TCP connection and unblocks the reader goroutine.
 			if cancelStream != nil {
 				cancelStream(errStreamIdle)
 			}
 			return streamResult{Stopped: true}, fmt.Errorf("model stream stalled after %s with no data: %w",
-				defaultStreamIdleTimeout, errStreamIdle)
+				idleTimeout, errStreamIdle)
 		}
 	}
 

pkg/runtime/streaming_test.go

@@ -67,7 +67,7 @@ func TestHandleStream_Refusal(t *testing.T) {
 	evCh := make(chan Event, 64)
 	res, err := handleStream(
 		t.Context(), nil, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout,
 	)
 	require.NoError(t, err)
 
@@ -95,7 +95,7 @@ func TestHandleStream_RefusalDropsPartialToolCalls(t *testing.T) {
 	evCh := make(chan Event, 64)
 	res, err := handleStream(
 		t.Context(), nil, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout,
 	)
 	require.NoError(t, err)
 
@@ -122,7 +122,7 @@ func TestHandleStream_ToolCallAndStopInSameChunk(t *testing.T) {
 	evCh := make(chan Event, 64) // buffered so handleStream never blocks on Emit
 	res, err := handleStream(
 		t.Context(), nil, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout,
 	)
 	require.NoError(t, err)
 
@@ -152,7 +152,7 @@ func TestHandleStream_ToolCallThenSeparateStop(t *testing.T) {
 	evCh := make(chan Event, 64)
 	res, err := handleStream(
 		t.Context(), nil, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout,
 	)
 	require.NoError(t, err)
 
@@ -181,7 +181,7 @@ func TestHandleStream_WhitespaceOnlyContentStops(t *testing.T) {
 	evCh := make(chan Event, 64)
 	res, err := handleStream(
 		t.Context(), nil, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), defaultStreamIdleTimeout,
 	)
 	require.NoError(t, err)
 
@@ -215,21 +215,12 @@ func (s *stalledStream) Close() {
 	s.closeOnce.Do(func() { close(s.unblock) })
 }
 
-// withShortStreamIdleTimeout temporarily overrides defaultStreamIdleTimeout
-// to shorten it for tests. Restores the original value via t.Cleanup.
-func withShortStreamIdleTimeout(t *testing.T, d time.Duration) {
-	t.Helper()
-	orig := defaultStreamIdleTimeout
-	defaultStreamIdleTimeout = d
-	t.Cleanup(func() { defaultStreamIdleTimeout = orig })
-}
-
 // TestHandleStream_IdleTimeout verifies that handleStream returns an error
 // wrapping errStreamIdle when no SSE chunk arrives within the idle window.
 // It also checks that the provided cancelStream function is called so the
 // HTTP transport can close the underlying TCP connection.
 func TestHandleStream_IdleTimeout(t *testing.T) {
-	withShortStreamIdleTimeout(t, 50*time.Millisecond)
+	t.Parallel()
 
 	stream := newStalledStream()
 	a := agent.New("root", "test", agent.WithModel(&mockProvider{id: "test/mock-model", stream: stream}))
@@ -244,7 +235,7 @@ func TestHandleStream_IdleTimeout(t *testing.T) {
 	evCh := make(chan Event, 64)
 	res, err := handleStream(
 		t.Context(), cancelStream, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), 50*time.Millisecond,
 	)
 
 	require.Error(t, err)
@@ -257,8 +248,7 @@ func TestHandleStream_IdleTimeout(t *testing.T) {
 // promptly when the caller's context is cancelled, even while a Recv call
 // is blocked. This covers the SIGTERM / graceful-shutdown path.
 func TestHandleStream_ContextCancellation(t *testing.T) {
-	// Use a long idle timeout so only context cancellation can trigger.
-	withShortStreamIdleTimeout(t, 10*time.Minute)
+	t.Parallel()
 
 	stream := newStalledStream()
 	a := agent.New("root", "test", agent.WithModel(&mockProvider{id: "test/mock-model", stream: stream}))
@@ -275,9 +265,10 @@ func TestHandleStream_ContextCancellation(t *testing.T) {
 
 	evCh := make(chan Event, 64)
 	_, cancelStream := context.WithCancelCause(ctx)
+	// Use a long idle timeout so only context cancellation can trigger.
 	res, err := handleStream(
 		ctx, cancelStream, stream, a, nil, sess, nil,
-		defaultTelemetry{}, NewChannelSink(evCh),
+		defaultTelemetry{}, NewChannelSink(evCh), 10*time.Minute,
 	)
 
 	require.Error(t, err)