feat: propagate proxy variables to jobs, services and builds (#1112)

Set `http_proxy`, `https_proxy` and `no_proxy` in the runner's environment and everything the runner controls uses them.

Go already read them for the runner's own requests. This adds jobs, in lower and upper case, service containers, and Dockerfile action builds.

Some hosts are added to `no_proxy` for jobs so they stay direct: the cache server, loopback, the job's service containers, and a `tcp://` Docker daemon. Without the last one the Docker client sends its API calls to the proxy and docker-in-docker breaks. Gitea is not added.

Images are pulled by the Docker daemon, which has its own proxy setting. In the `dind` images it reads these same variables. The runner warns at startup if it has a proxy and the daemon does not.

Fixes https://gitea.com/gitea/runner/issues/1118, originally reported as https://gitea.com/gitea/runner/issues/708.

---------

Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com>
Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1112
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
Co-authored-by: bircni <bircni@icloud.com>

.golangci.yml

@@ -11,6 +11,7 @@ linters:
     - dupl
     - errcheck
     - forbidigo
+    - forcetypeassert
     - gocheckcompilerdirectives
     - gocritic
     - goheader
@@ -102,6 +103,9 @@ linters:
       - linters:
           - forbidigo
         path: cmd
+      - linters:
+          - forcetypeassert
+        path: _test\.go
 issues:
   max-issues-per-linter: 0
   max-same-issues: 0

README.md

@@ -209,6 +209,35 @@ Whenever the resulting labels differ from the ones in the registration file, the
 
 > **Note:** A runner that only exposes `host` labels still needs access to a Docker daemon (e.g. a mounted `/var/run/docker.sock`) whenever a job uses a `docker://` action or a service container. `host` labels only change where the job's own steps run; container-based steps and actions are still executed with Docker.
 
+#### Proxy
+
+Set these variables in the runner's environment, with systemd `Environment=`, `docker run -e`, or Kubernetes `env:`:
+
+```sh
+http_proxy=http://proxy.example:3128
+https_proxy=http://proxy.example:3128
+no_proxy=gitea.internal,.example.local
+```
+
+The runner uses them for its own requests and gives them to every job, in lower and upper case.
+
+These hosts are added to `no_proxy` for jobs, so they are always reached directly:
+
+- the cache server
+- `localhost`, `127.0.0.1` and `::1`
+- the job's service containers
+- the Docker daemon, when it is reached over `tcp://`
+
+Gitea is not added. Add it to `no_proxy` yourself if it should be reached directly.
+
+To change a value for one job, set it in a step's `env:` or in the job's `container.env`. Setting it at workflow or job level has no effect. To change it for the whole runner, set it in `runner.envs`. A `no_proxy` set there is added to the list above instead of replacing it.
+
+Images are pulled by the Docker daemon, which needs its own proxy setting. In the `dind` images the daemon runs in the same container and reads the variables above. For any other daemon, see [the Docker documentation](https://docs.docker.com/engine/daemon/proxy/). The runner logs a warning at startup if it has a proxy and the daemon does not.
+
+Dockerfile actions are built with these variables as build arguments, so their `RUN` steps can reach the network.
+
+A password in a proxy URL is hidden in job logs. Any step can still read it, because the step is given the proxy URL in its environment.
+
 #### Caching (`actions/cache`)
 
 Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default.

act/artifactcache/handler.go

@@ -70,6 +70,7 @@ type Handler struct {
 	storage  *Storage
 	router   *httprouter.Router
 	listener net.Listener
+	port     int
 	server   *http.Server
 	logger   logrus.FieldLogger
 
@@ -177,6 +178,12 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
 	if err != nil {
 		return nil, err
 	}
+	addr, ok := listener.Addr().(*net.TCPAddr)
+	if !ok {
+		listener.Close()
+		return nil, fmt.Errorf("cache server listens on %T, want a TCP address", listener.Addr())
+	}
+	h.port = addr.Port
 	server := &http.Server{
 		ReadHeaderTimeout: 2 * time.Second,
 		Handler:           router,
@@ -194,9 +201,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
 
 func (h *Handler) ExternalURL() string {
 	// TODO: make the external url configurable if necessary
-	return fmt.Sprintf("http://%s:%d",
-		h.outboundIP,
-		h.listener.Addr().(*net.TCPAddr).Port)
+	return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
 }
 
 // RegisterJob makes token a valid bearer credential for cache requests from

act/common/git/git.go

@@ -16,6 +16,7 @@ import (
 	"sync"
 
 	"gitea.com/gitea/runner/act/common"
+	"gitea.com/gitea/runner/internal/pkg/lock"
 
 	"github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/config"
@@ -32,7 +33,7 @@ var (
 	githubHTTPRegex     = regexp.MustCompile(`^https?://.*github.com.*/(.+)/(.+?)(?:.git)?$`)
 	githubSSHRegex      = regexp.MustCompile(`github.com[:/](.+)/(.+?)(?:.git)?$`)
 
-	cloneLocks sync.Map // key: clone target directory; value: *sync.Mutex
+	cloneLocks lock.Keyed[string] // key: clone target directory
 
 	ErrShortRef = errors.New("short SHA references are not supported")
 	ErrNoRepo   = errors.New("unable to find git repo")
@@ -43,10 +44,7 @@ var (
 // Callers reading files inside dir (e.g. tarring a checked-out action into a job container) must hold this lock too,
 // otherwise a concurrent NewGitCloneExecutor on the same dir can mutate the worktree mid-read.
 func AcquireCloneLock(dir string) func() {
-	v, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
-	mu := v.(*sync.Mutex)
-	mu.Lock()
-	return mu.Unlock
+	return cloneLocks.Lock(dir)
 }
 
 type Error struct {

act/common/git/git_test.go

@@ -12,7 +12,6 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
-	"sync"
 	"syscall"
 	"testing"
 	"time"
@@ -610,12 +609,4 @@ func TestAcquireCloneLock(t *testing.T) {
 			t.Fatal("acquire on a different directory must not block")
 		}
 	})
-
-	t.Run("same directory reuses the same mutex", func(t *testing.T) {
-		dir := t.TempDir()
-
-		v1, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
-		v2, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
-		require.Same(t, v1, v2)
-	})
 }

act/common/outbound_ip.go

@@ -19,7 +19,9 @@ func GetOutboundIP() net.IP {
 	conn, err := net.Dial("udp", "8.8.8.8:80")
 	if err == nil {
 		defer conn.Close()
-		return conn.LocalAddr().(*net.UDPAddr).IP
+		if addr, ok := conn.LocalAddr().(*net.UDPAddr); ok {
+			return addr.IP
+		}
 	}
 
 	// So the machine cannot access the internet. Pick an IP address from network interfaces.

act/container/container_types.go

@@ -82,6 +82,7 @@ type NewDockerBuildExecutorInput struct {
 	BuildContext io.Reader
 	ImageTag     string
 	Platform     string
+	BuildArgs    map[string]*string
 }
 
 // NewDockerNetworkCreateExecutorInput the input for the NewDockerNetworkCreateExecutor function

act/container/docker_build.go

@@ -49,6 +49,7 @@ func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
 			Remove:      true,
 			AuthConfigs: LoadDockerAuthConfigs(ctx),
 			Dockerfile:  input.Dockerfile,
+			BuildArgs:   input.BuildArgs,
 		}
 		platform, err := parsePlatform(input.Platform)
 		if err != nil {

act/container/docker_create_flags_test.go

@@ -51,7 +51,8 @@ func TestCreateFlagsValidate(t *testing.T) {
 
 func TestNewContainerAppliesCreateFlags(t *testing.T) {
 	input := &NewContainerInput{Platform: "linux/amd64", Options: "--platform linux/arm64 --pull never"}
-	cr := NewContainer(input).(*containerReference)
+	cr, ok := NewContainer(input).(*containerReference)
+	require.True(t, ok)
 	assert.Equal(t, "linux/arm64", input.Platform)
 	assert.Equal(t, pullPolicyNever, cr.pullPolicy)
 

act/exprparser/interpreter_test.go

@@ -11,6 +11,7 @@ import (
 	"gitea.com/gitea/runner/act/model"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestLiterals(t *testing.T) {
@@ -523,7 +524,9 @@ func TestOperatorsBooleanEvaluation(t *testing.T) {
 			assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
 
 			if expected, ok := tt.expected.(float64); ok && math.IsNaN(expected) {
-				assert.True(t, math.IsNaN(output.(float64)))
+				number, ok := output.(float64)
+				require.True(t, ok, "want a number, got %T", output)
+				assert.True(t, math.IsNaN(number))
 			} else {
 				assert.Equal(t, tt.expected, output)
 			}

act/model/workflow.go

@@ -86,11 +86,12 @@ func (w *Workflow) OnSchedule() []string {
 	case []any:
 		allSchedules := []string{}
 		for _, v := range val {
-			for k, cron := range v.(map[string]any) {
-				if k != "cron" {
-					continue
-				}
-				allSchedules = append(allSchedules, cron.(string))
+			entry, ok := v.(map[string]any)
+			if !ok {
+				continue
+			}
+			if cron, ok := entry["cron"].(string); ok {
+				allSchedules = append(allSchedules, cron)
 			}
 		}
 		return allSchedules
@@ -443,9 +444,9 @@ func normalizeMatrixValue(key string, val any) ([]any, error) {
 // Scalar values are wrapped into single-element arrays automatically.
 // Template expressions are resolved by EvaluateYamlNode before this method is
 // called; if unresolved, the literal string is wrapped as a one-element fallback.
-func (j *Job) Matrix() map[string][]any {
+func (j *Job) Matrix() (map[string][]any, error) {
 	if j.Strategy == nil || j.Strategy.RawMatrix.Kind != yaml.MappingNode {
-		return nil
+		return map[string][]any{}, nil
 	}
 
 	// Decode to flexible map first so that scalar values don't cause a type error.
@@ -455,9 +456,9 @@ func (j *Job) Matrix() map[string][]any {
 		// Fall back to the strict array-only format for backward compatibility.
 		var val map[string][]any
 		if !decodeNode(j.Strategy.RawMatrix, &val) {
-			return nil
+			return map[string][]any{}, nil
 		}
-		return val
+		return val, nil
 	}
 
 	// Convert flexible format to expected format with validation
@@ -465,12 +466,11 @@ func (j *Job) Matrix() map[string][]any {
 	for k, v := range flexVal {
 		normalized, err := normalizeMatrixValue(k, v)
 		if err != nil {
-			log.Errorf("matrix validation error: %v", err)
-			return nil
+			return nil, err
 		}
 		val[k] = normalized
 	}
-	return val
+	return val, nil
 }
 
 // GetMatrixes returns the matrix cross product
@@ -482,38 +482,38 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) {
 		j.Strategy.FailFast = j.Strategy.GetFailFast()
 		j.Strategy.MaxParallel = j.Strategy.GetMaxParallel()
 
-		if m := j.Matrix(); m != nil {
+		m, err := j.Matrix()
+		if err != nil {
+			return nil, err
+		}
+		if len(m) > 0 {
 			includes := make([]map[string]any, 0)
 			extraIncludes := make([]map[string]any, 0)
+			addInclude := func(raw any) error {
+				include, ok := raw.(map[string]any)
+				if !ok {
+					return fmt.Errorf("the workflow is not valid. Matrix include %v is not a map of matrix keys to values", raw)
+				}
+				for k := range include {
+					if _, ok := m[k]; ok {
+						includes = append(includes, include)
+						return nil
+					}
+				}
+				extraIncludes = append(extraIncludes, include)
+				return nil
+			}
 			for _, v := range m["include"] {
 				switch t := v.(type) {
 				case []any:
 					for _, i := range t {
-						i := i.(map[string]any)
-						extraInclude := true
-						for k := range i {
-							if _, ok := m[k]; ok {
-								includes = append(includes, i)
-								extraInclude = false
-								break
-							}
-						}
-						if extraInclude {
-							extraIncludes = append(extraIncludes, i)
+						if err := addInclude(i); err != nil {
+							return nil, err
 						}
 					}
 				case any:
-					v := v.(map[string]any)
-					extraInclude := true
-					for k := range v {
-						if _, ok := m[k]; ok {
-							includes = append(includes, v)
-							extraInclude = false
-							break
-						}
-					}
-					if extraInclude {
-						extraIncludes = append(extraIncludes, v)
+					if err := addInclude(t); err != nil {
+						return nil, err
 					}
 				}
 			}
@@ -521,10 +521,13 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) {
 
 			excludes := make([]map[string]any, 0)
 			for _, e := range m["exclude"] {
-				e := e.(map[string]any)
-				for k := range e {
+				exclude, ok := e.(map[string]any)
+				if !ok {
+					return nil, fmt.Errorf("the workflow is not valid. Matrix exclude %v is not a map of matrix keys to values", e)
+				}
+				for k := range exclude {
 					if _, ok := m[k]; ok {
-						excludes = append(excludes, e)
+						excludes = append(excludes, exclude)
 					} else {
 						// We fail completely here because that's what GitHub does for non-existing matrix keys, fail on exclude, silent skip on include
 						return nil, fmt.Errorf("the workflow is not valid. Matrix exclude key %q does not match any key within the matrix", k)

act/model/workflow_test.go

@@ -667,7 +667,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
 	matrixes, err := job.GetMatrixes()
 	assert.NoError(t, err)                          //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
-	assert.Equal(t, job.Matrix(), map[string][]any(nil))
+	matrix, err := job.Matrix()
+	require.NoError(t, err)
+	assert.Empty(t, matrix)
 	assert.Equal(t, job.Strategy.MaxParallel, 2) //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, job.Strategy.FailFast, true) //nolint:testifylint // pre-existing issue from nektos/act
 
@@ -675,7 +677,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
 	matrixes, err = job.GetMatrixes()
 	assert.NoError(t, err)                          //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
-	assert.Equal(t, job.Matrix(), map[string][]any(nil))
+	matrix, err = job.Matrix()
+	require.NoError(t, err)
+	assert.Empty(t, matrix)
 	assert.Equal(t, job.Strategy.MaxParallel, 4)  //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act
 
@@ -683,7 +687,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
 	matrixes, err = job.GetMatrixes()
 	assert.NoError(t, err)                          //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
-	assert.Equal(t, job.Matrix(), map[string][]any(nil))
+	matrix, err = job.Matrix()
+	require.NoError(t, err)
+	assert.Empty(t, matrix)
 	assert.Equal(t, job.Strategy.MaxParallel, 2)  //nolint:testifylint // pre-existing issue from nektos/act
 	assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act
 
@@ -700,7 +706,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
 			{"datacenter": "site-b", "node-version": "12.x", "site": "dev"},
 		},
 	)
-	assert.Equal(t, job.Matrix(), //nolint:testifylint // pre-existing issue from nektos/act
+	matrix, err = job.Matrix()
+	require.NoError(t, err)
+	assert.Equal(t, matrix, //nolint:testifylint // pre-existing issue from nektos/act
 		map[string][]any{
 			"datacenter": {"site-c", "site-d"},
 			"exclude": {
@@ -1092,13 +1100,15 @@ jobs:
 				t.Fatal("job not found")
 			}
 
-			matrix := job.Matrix()
+			matrix, err := job.Matrix()
 
 			if tt.wantErr {
+				require.Error(t, err)
 				assert.Nil(t, matrix, "matrix should be nil on error")
 			} else {
+				require.NoError(t, err)
 				if tt.wantLen == 0 {
-					assert.Nil(t, matrix, "matrix should be nil for jobs without strategy")
+					assert.Empty(t, matrix, "no matrix for jobs without strategy")
 				} else {
 					assert.NotNil(t, matrix, "matrix should not be nil")
 					assert.Len(t, matrix, tt.wantLen, "matrix should have expected number of keys")
@@ -1130,11 +1140,9 @@ func TestJobMatrixValidation(t *testing.T) {
 			},
 		}
 
-		// Attempt to get matrix
-		matrix := job.Matrix()
-
-		// Should return nil due to validation error
-		assert.Nil(t, matrix, "matrix with nested map should return nil")
+		matrix, err := job.Matrix()
+		require.ErrorContains(t, err, `matrix key "config" has invalid nested object value`)
+		assert.Nil(t, matrix)
 	})
 }
 

act/runner/action.go

@@ -129,6 +129,16 @@ func readActionImpl(ctx context.Context, step *model.Step, actionDir, actionPath
 	return action, err
 }
 
+// cachedActionTar returns the action's tree from the action cache, which only a remote action
+// has an entry in.
+func cachedActionTar(ctx context.Context, step actionStep, name, includePrefix string) (io.ReadCloser, error) {
+	remote, ok := step.(*stepActionRemote)
+	if !ok {
+		return nil, fmt.Errorf("action %q is a remote action but runs as %T", name, step)
+	}
+	return step.getRunContext().Config.ActionCache.GetTarArchive(ctx, remote.cacheDir, remote.resolvedSha, includePrefix)
+}
+
 func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actionPath, containerActionDir string) error {
 	logger := common.Logger(ctx)
 	rc := step.getRunContext()
@@ -147,8 +157,7 @@ func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actio
 	}
 
 	if rc.Config != nil && rc.Config.ActionCache != nil {
-		raction := step.(*stepActionRemote)
-		ta, err := rc.Config.ActionCache.GetTarArchive(ctx, raction.cacheDir, raction.resolvedSha, "")
+		ta, err := cachedActionTar(ctx, step, stepModel.Uses, "")
 		if err != nil {
 			return err
 		}
@@ -351,8 +360,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
 				}
 				defer buildContext.Close()
 			} else if rc.Config.ActionCache != nil {
-				rstep := step.(*stepActionRemote)
-				buildContext, err = rc.Config.ActionCache.GetTarArchive(ctx, rstep.cacheDir, rstep.resolvedSha, contextDir)
+				buildContext, err = cachedActionTar(ctx, step, actionName, contextDir)
 				if err != nil {
 					return err
 				}
@@ -364,6 +372,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
 				ImageTag:     image,
 				BuildContext: buildContext,
 				Platform:     rc.Config.ContainerArchitecture,
+				BuildArgs:    rc.proxyBuildArgs(),
 			})
 			if buildContext == nil {
 				// Held across the whole build: the daemon drains contextDir lazily.

act/runner/run_context.go

@@ -22,13 +22,13 @@ import (
 	"runtime"
 	"slices"
 	"strings"
-	"sync"
 	"time"
 
 	"gitea.com/gitea/runner/act/common"
 	"gitea.com/gitea/runner/act/container"
 	"gitea.com/gitea/runner/act/exprparser"
 	"gitea.com/gitea/runner/act/model"
+	"gitea.com/gitea/runner/internal/pkg/lock"
 
 	"github.com/docker/cli/cli/compose/loader"
 	"github.com/docker/go-connections/nat"
@@ -435,7 +435,9 @@ func (rc *RunContext) startJobContainer() common.Executor {
 				continue
 			}
 			// interpolate env
-			interpolatedEnvs := make(map[string]string, len(spec.Env))
+			interpolatedEnvs := make(map[string]string, len(spec.Env)+len(rc.Config.ProxyEnv))
+			// a service reaches the internet the way the job does; its own env still wins
+			maps0.Copy(interpolatedEnvs, rc.Config.ProxyEnv)
 			for k, v := range spec.Env {
 				interpolatedEnvs[k] = rc.ExprEval.Interpolate(ctx, v)
 			}
@@ -715,13 +717,10 @@ func (rc *RunContext) ActionCacheDir() string {
 // jobMutexes serializes per-job result/output aggregation across the matrix combinations that
 // share one *model.Job and run in parallel. Keyed by the shared *model.Job (mirrors the
 // per-directory AcquireCloneLock pattern).
-var jobMutexes sync.Map // key: *model.Job; value: *sync.Mutex
+var jobMutexes lock.Keyed[*model.Job]
 
 func lockJob(job *model.Job) func() {
-	v, _ := jobMutexes.LoadOrStore(job, &sync.Mutex{})
-	mu := v.(*sync.Mutex)
-	mu.Lock()
-	return mu.Unlock
+	return jobMutexes.Lock(job)
 }
 
 func (rc *RunContext) interpolateOutputs() common.Executor {
@@ -972,6 +971,20 @@ func (rc *RunContext) isEnabled(ctx context.Context) (bool, error) {
 	return true, nil
 }
 
+// proxyBuildArgs returns the job's proxy variables as docker build args. The docker CLI
+// pre-populates these from its own client configuration, but act builds through the API,
+// so without them a Dockerfile action's RUN steps have no network behind a proxy.
+func (rc *RunContext) proxyBuildArgs() map[string]*string {
+	if len(rc.Config.ProxyEnv) == 0 {
+		return nil
+	}
+	args := make(map[string]*string, len(rc.Config.ProxyEnv))
+	for name, value := range rc.Config.ProxyEnv {
+		args[name] = &value
+	}
+	return args
+}
+
 func mergeMaps(maps ...map[string]string) map[string]string {
 	rtnMap := make(map[string]string)
 	for _, m := range maps {

act/runner/run_context_test.go

@@ -292,6 +292,83 @@ jobs:
 	require.Equal(t, [2]string{"", ""}, credentials["redis:latest"])
 }
 
+// A service container reaches the internet the same way the job does, so it inherits the
+// job's proxy; a service that sets the variable itself keeps its own value.
+func TestStartJobContainerGivesServicesTheJobProxy(t *testing.T) {
+	workflow, err := model.ReadWorkflow(strings.NewReader(`
+name: test
+on: push
+jobs:
+  job:
+    runs-on: ubuntu-latest
+    container:
+      image: registry.example/job:latest
+    services:
+      redis:
+        image: redis:latest
+      db:
+        image: postgres:latest
+        env:
+          no_proxy: db-only.example
+    steps: []
+`))
+	require.NoError(t, err)
+
+	var inputs []*container.NewContainerInput
+	origNewContainer := newContainer
+	newContainer = func(input *container.NewContainerInput) container.ExecutionsEnvironment {
+		inputs = append(inputs, input)
+		return fakeContainer{}
+	}
+	t.Cleanup(func() { newContainer = origNewContainer })
+
+	rc := &RunContext{
+		Name: "test",
+		Config: &Config{
+			Workdir:              "/tmp",
+			ContainerNetworkMode: "host",
+			ReuseContainers:      true,
+			Env:                  map[string]string{},
+			ProxyEnv:             map[string]string{"http_proxy": "http://proxy:3128", "no_proxy": "internal.example"},
+			Secrets:              map[string]string{},
+		},
+		Env: map[string]string{},
+		Run: &model.Run{
+			JobID:    "job",
+			Workflow: workflow,
+		},
+	}
+	rc.ExprEval = rc.NewExpressionEvaluator(t.Context())
+
+	require.NoError(t, rc.startJobContainer()(t.Context()))
+
+	env := map[string][]string{}
+	for _, in := range inputs {
+		env[in.Image] = in.Env
+	}
+
+	require.Contains(t, env["redis:latest"], "http_proxy=http://proxy:3128")
+	require.Contains(t, env["redis:latest"], "no_proxy=internal.example")
+	// the service's own env wins over what the runner injected, without dropping the rest
+	require.Contains(t, env["postgres:latest"], "no_proxy=db-only.example")
+	require.NotContains(t, env["postgres:latest"], "no_proxy=internal.example")
+	require.Contains(t, env["postgres:latest"], "http_proxy=http://proxy:3128")
+}
+
+// act builds Dockerfile actions through the API, which does not pre-populate the proxy
+// build args the docker CLI would, so the RUN steps would have no network behind a proxy.
+func TestProxyBuildArgs(t *testing.T) {
+	rc := &RunContext{Config: &Config{ProxyEnv: map[string]string{"http_proxy": "http://proxy:3128"}}}
+
+	args := rc.proxyBuildArgs()
+
+	require.Len(t, args, 1)
+	require.Equal(t, "http://proxy:3128", *args["http_proxy"])
+
+	// a job without a proxy builds exactly as it does today
+	require.Nil(t, (&RunContext{Config: &Config{}}).proxyBuildArgs())
+}
+
 func TestRunContext_GetBindsAndMounts(t *testing.T) {
 	rctemplate := &RunContext{
 		Name: "TestRCName",

act/runner/runner.go

@@ -73,6 +73,7 @@ type Config struct {
 	ContainerNetworkMode               docker_container.NetworkMode                  // the network mode of job containers (the value of --network)
 	ContainerNetworkCreateOptions      container.NewDockerNetworkCreateExecutorInput // the default network create options
 	ActionCache                        ActionCache                                   // Use a custom ActionCache Implementation
+	ProxyEnv                           map[string]string                             // the proxy variables the job runs with, also given to service containers and image builds
 
 	PresetGitHubContext   *model.GithubContext // the preset github context, overrides some fields like DefaultBranch, Env, Secrets etc.
 	EventJSON             string               // the content of JSON file to use for event.json in containers, overrides EventPath

act/runner/step_action_remote.go

@@ -138,9 +138,10 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
 		})
 		var ntErr common.Executor
 		if err := gitClone(ctx); err != nil {
-			if errors.Is(err, git.ErrShortRef) {
+			var refErr *git.Error
+			if errors.As(err, &refErr) && errors.Is(err, git.ErrShortRef) {
 				return fmt.Errorf("Unable to resolve action `%s`, the provided ref `%s` is the shortened version of a commit SHA, which is not supported. Please use the full commit SHA `%s` instead",
-					sar.Step.Uses, sar.remoteAction.Ref, err.(*git.Error).Commit())
+					sar.Step.Uses, sar.remoteAction.Ref, refErr.Commit())
 			} else if errors.Is(err, gogit.ErrForceNeeded) { // TODO: figure out if it will be easy to shadow/alias go-git err's
 				ntErr = common.NewInfoExecutor("Non-terminating error while running 'git clone': %v", err)
 			} else {

go.mod

@@ -38,6 +38,7 @@ require (
 	github.com/timshannon/bolthold v0.0.0-20240314194003-30aac6950928
 	go.etcd.io/bbolt v1.5.0
 	go.yaml.in/yaml/v4 v4.0.0-rc.3
+	golang.org/x/net v0.56.0
 	golang.org/x/sys v0.47.0
 	golang.org/x/term v0.45.0
 	golang.org/x/text v0.40.0
@@ -105,7 +106,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.44.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.53.0 // indirect
-	golang.org/x/net v0.56.0 // indirect
 	golang.org/x/sync v0.22.0 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

internal/app/cmd/daemon.go

@@ -64,6 +64,14 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
 			log.Warn("no labels configured, runner may not be able to pick up jobs")
 		}
 
+		// Before the first Docker API call: the standard library resolves the proxy
+		// environment once. Ungated because host labels still reach the daemon.
+		if dockerSocketPath, err := getDockerSocketPath(cfg.Container.DockerHost); err == nil {
+			run.BypassProxyForDockerHost(dockerSocketPath)
+		} else {
+			log.Debugf("cannot resolve the docker socket path, so Docker API calls are not exempted from the proxy: %v", err)
+		}
+
 		if ls.RequireDocker() || cfg.Container.RequireDocker {
 			// Wait for dockerd be ready
 			if timeout := cfg.Container.DockerTimeout; timeout > 0 {
@@ -101,6 +109,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
 			}
 			// if dockerSocketPath passes the check, override DOCKER_HOST with dockerSocketPath
 			os.Setenv("DOCKER_HOST", dockerSocketPath)
+			run.WarnIfDaemonHasNoProxy(ctx)
 			// empty cfg.Container.DockerHost means runner need to find an available docker host automatically
 			// and assign the path to cfg.Container.DockerHost
 			if cfg.Container.DockerHost == "" {

internal/app/cmd/exec.go

@@ -22,6 +22,7 @@ import (
 	"gitea.com/gitea/runner/act/common"
 	"gitea.com/gitea/runner/act/model"
 	"gitea.com/gitea/runner/act/runner"
+	"gitea.com/gitea/runner/internal/app/run"
 
 	"github.com/joho/godotenv"
 	"github.com/moby/moby/api/types/container"
@@ -415,6 +416,11 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
 		}
 		handler.RegisterJob(actionsRuntimeToken, "__local/__exec")
 
+		// no service aliases: exec builds one config for the whole plan
+		run.BypassProxyForDockerHost(os.Getenv("DOCKER_HOST"))
+		proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil)
+		maps.Copy(env, proxyEnv)
+
 		// run the plan
 		config := &runner.Config{
 			Workdir:               execArgs.Workdir(),
@@ -425,6 +431,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
 			LogOutput:             true,
 			JSONLogger:            execArgs.jsonLogger,
 			Env:                   env,
+			ProxyEnv:              proxyEnv,
 			Vars:                  execArgs.LoadVars(),
 			Secrets:               execArgs.LoadSecrets(),
 			InsecureSecrets:       execArgs.insecureSecrets,

internal/app/run/proxy.go

@@ -0,0 +1,163 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package run
+
+import (
+	"context"
+	"net/url"
+	"os"
+	"slices"
+	"strings"
+
+	"gitea.com/gitea/runner/act/container"
+
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/net/http/httpproxy"
+)
+
+// proxyFromEnv returns the runner's own proxy configuration, or nil when it has none.
+func proxyFromEnv() *httpproxy.Config {
+	cfg := httpproxy.FromEnvironment()
+	if cfg.HTTPProxy == "" && cfg.HTTPSProxy == "" {
+		return nil
+	}
+	return cfg
+}
+
+// JobProxyEnv returns the proxy variables a job runs with, given what runner.envs already
+// put in jobEnvs. Gitea is deliberately not made direct, so it stays reachable however the
+// runner reaches it.
+func JobProxyEnv(jobEnvs map[string]string, cacheURL string, serviceNames []string) map[string]string {
+	cfg := proxyFromEnv()
+	if cfg == nil {
+		return nil
+	}
+
+	// Go bypasses loopback on its own, curl and most other tools in a job do not.
+	direct := append([]string{"localhost", "127.0.0.1", "::1"}, serviceNames...)
+	direct = append(direct, hostOf(cacheURL))
+
+	proxyEnv := map[string]string{}
+	setPair := func(lower, upper, value string) {
+		if value == "" {
+			return
+		}
+		// Either spelling in runner.envs takes over both, so the pair cannot disagree.
+		if existing, ok := jobEnvs[lower]; ok {
+			value = existing
+		} else if existing, ok := jobEnvs[upper]; ok {
+			value = existing
+		}
+		proxyEnv[lower], proxyEnv[upper] = value, value
+	}
+	setPair("http_proxy", "HTTP_PROXY", cfg.HTTPProxy)
+	setPair("https_proxy", "HTTPS_PROXY", cfg.HTTPSProxy)
+
+	// no_proxy is merged rather than replaced: the hosts above are structural, and an
+	// operator cannot list the cache server's startup-assigned address in advance.
+	noProxy := appendNoProxy(cfg.NoProxy, direct...)
+	for _, name := range []string{"no_proxy", "NO_PROXY"} {
+		if existing, ok := jobEnvs[name]; ok {
+			noProxy = appendNoProxy(existing, strings.Split(noProxy, ",")...)
+			break
+		}
+	}
+	proxyEnv["no_proxy"], proxyEnv["NO_PROXY"] = noProxy, noProxy
+	return proxyEnv
+}
+
+// BypassProxyForDockerHost keeps the runner's Docker API traffic off the proxy: the docker
+// client proxies every transport that is not a unix socket or a named pipe, so a tcp://
+// daemon would be reached through a proxy that cannot route to it.
+//
+// It must run before the first Docker API call, because the standard library resolves the
+// proxy environment once.
+func BypassProxyForDockerHost(dockerHost string) {
+	cfg := proxyFromEnv()
+	if cfg == nil {
+		return
+	}
+	host := hostOf(dockerHost)
+	if host == "" {
+		// A unix socket or named pipe is never proxied.
+		return
+	}
+
+	noProxy := appendNoProxy(cfg.NoProxy, host)
+	for _, name := range []string{"no_proxy", "NO_PROXY"} {
+		if err := os.Setenv(name, noProxy); err != nil {
+			log.Warnf("cannot set %s for the runner process: %v", name, err)
+		}
+	}
+	log.Debugf("docker host %s is reached directly, no_proxy is now %q", host, noProxy)
+}
+
+// WarnIfDaemonHasNoProxy points at the one part the runner cannot set: the docker daemon
+// pulls the images, and a daemon in its own container needs its own proxy.
+func WarnIfDaemonHasNoProxy(ctx context.Context) {
+	if proxyFromEnv() == nil {
+		return
+	}
+	info, err := container.GetHostInfo(ctx)
+	if err != nil {
+		log.Debugf("cannot read the docker daemon's proxy configuration: %v", err)
+		return
+	}
+	if info.HTTPProxy == "" && info.HTTPSProxy == "" {
+		log.Warn("the runner has a proxy but the docker daemon reports none, so image pulls will not use it: https://docs.docker.com/engine/daemon/proxy/")
+	}
+}
+
+// proxyPasswords returns the passwords embedded in the runner's proxy URLs, to mask before
+// a job echoes its environment.
+func proxyPasswords() []string {
+	cfg := proxyFromEnv()
+	if cfg == nil {
+		return nil
+	}
+
+	var passwords []string
+	for _, raw := range []string{cfg.HTTPProxy, cfg.HTTPSProxy} {
+		parsed, err := url.Parse(raw)
+		if err != nil || parsed.User == nil {
+			continue
+		}
+		if password, ok := parsed.User.Password(); ok && password != "" && !slices.Contains(passwords, password) {
+			passwords = append(passwords, password)
+		}
+	}
+	return passwords
+}
+
+// appendNoProxy adds hosts to a no_proxy list, keeping the operator's entries and adding
+// none twice.
+func appendNoProxy(noProxy string, hosts ...string) string {
+	entries := []string{}
+	for entry := range strings.SplitSeq(noProxy, ",") {
+		if entry = strings.TrimSpace(entry); entry != "" {
+			entries = append(entries, entry)
+		}
+	}
+
+	for _, host := range hosts {
+		if host == "" || slices.Contains(entries, host) {
+			continue
+		}
+		entries = append(entries, host)
+	}
+	return strings.Join(entries, ",")
+}
+
+// hostOf returns the host of a URL without its port, the form a no_proxy entry takes. It is
+// empty for anything without a network host, such as a unix socket.
+func hostOf(raw string) string {
+	if raw == "" {
+		return ""
+	}
+	parsed, err := url.Parse(strings.TrimSuffix(raw, "/"))
+	if err != nil || parsed.Hostname() == "" {
+		return ""
+	}
+	return parsed.Hostname()
+}

internal/app/run/proxy_test.go

@@ -0,0 +1,192 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package run
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// clearProxyEnv starts a test from a runner that has no proxy, whatever the developer's own
+// environment looks like.
+func clearProxyEnv(t *testing.T) {
+	t.Helper()
+	for _, name := range []string{"http_proxy", "HTTP_PROXY", "https_proxy", "HTTPS_PROXY", "no_proxy", "NO_PROXY"} {
+		t.Setenv(name, "")
+	}
+}
+
+func TestJobProxyEnv(t *testing.T) {
+	const loopback = "localhost,127.0.0.1,::1"
+	const proxy = "http://proxy:3128"
+
+	for _, tc := range []struct {
+		name     string
+		runner   map[string]string // the runner process environment
+		envs     map[string]string // what runner.envs already put in the job
+		cacheURL string
+		services []string
+		want     map[string]string
+	}{
+		{
+			// The guarantee that makes this safe to ship: a runner without a proxy gives its
+			// jobs nothing at all.
+			name:     "runner has no proxy",
+			cacheURL: "http://cache.local:8088/",
+		},
+		{
+			name:   "a lone no_proxy is not a proxy",
+			runner: map[string]string{"no_proxy": "example.com"},
+		},
+		{
+			name:   "both spellings of every variable",
+			runner: map[string]string{"http_proxy": proxy, "https_proxy": proxy},
+			want: map[string]string{
+				"http_proxy": proxy, "HTTP_PROXY": proxy,
+				"https_proxy": proxy, "HTTPS_PROXY": proxy,
+				"no_proxy": loopback, "NO_PROXY": loopback,
+			},
+		},
+		{
+			name:   "a variable the runner does not have is omitted",
+			runner: map[string]string{"https_proxy": proxy},
+			want: map[string]string{
+				"https_proxy": proxy, "HTTPS_PROXY": proxy,
+				"no_proxy": loopback, "NO_PROXY": loopback,
+			},
+		},
+		{
+			// The cache server and the service containers are on the local network; Gitea is
+			// not added and keeps being reached through the proxy.
+			name:     "hosts the job must reach directly",
+			runner:   map[string]string{"http_proxy": proxy, "no_proxy": "internal.example"},
+			cacheURL: "http://192.168.1.10:34567/",
+			services: []string{"postgres", "redis"},
+			want: map[string]string{
+				"http_proxy": proxy, "HTTP_PROXY": proxy,
+				"no_proxy": "internal.example," + loopback + ",postgres,redis,192.168.1.10",
+				"NO_PROXY": "internal.example," + loopback + ",postgres,redis,192.168.1.10",
+			},
+		},
+		{
+			name:   "runner.envs wins, for both spellings",
+			runner: map[string]string{"http_proxy": "http://from-env:3128"},
+			envs:   map[string]string{"http_proxy": "http://from-config:3128"},
+			want: map[string]string{
+				"http_proxy": "http://from-config:3128", "HTTP_PROXY": "http://from-config:3128",
+				"no_proxy": loopback, "NO_PROXY": loopback,
+			},
+		},
+		{
+			name:   "runner.envs wins through the upper case spelling too",
+			runner: map[string]string{"http_proxy": "http://from-env:3128"},
+			envs:   map[string]string{"HTTP_PROXY": "http://from-config:3128"},
+			want: map[string]string{
+				"http_proxy": "http://from-config:3128", "HTTP_PROXY": "http://from-config:3128",
+				"no_proxy": loopback, "NO_PROXY": loopback,
+			},
+		},
+		{
+			// A runner.envs no_proxy adds to the hosts that must stay direct rather than
+			// replacing them, which would send cache traffic through the proxy.
+			name:     "runner.envs no_proxy is merged, not substituted",
+			runner:   map[string]string{"http_proxy": proxy},
+			envs:     map[string]string{"no_proxy": "operator.example"},
+			cacheURL: "http://192.168.1.10:34567/",
+			want: map[string]string{
+				"http_proxy": proxy, "HTTP_PROXY": proxy,
+				"no_proxy": "operator.example," + loopback + ",192.168.1.10",
+				"NO_PROXY": "operator.example," + loopback + ",192.168.1.10",
+			},
+		},
+		{
+			name:     "runner.envs NO_PROXY is merged through the upper case spelling too",
+			runner:   map[string]string{"http_proxy": proxy},
+			envs:     map[string]string{"NO_PROXY": "operator.example"},
+			cacheURL: "http://192.168.1.10:34567/",
+			want: map[string]string{
+				"http_proxy": proxy, "HTTP_PROXY": proxy,
+				"no_proxy": "operator.example," + loopback + ",192.168.1.10",
+				"NO_PROXY": "operator.example," + loopback + ",192.168.1.10",
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			clearProxyEnv(t)
+			for name, value := range tc.runner {
+				t.Setenv(name, value)
+			}
+			assert.Equal(t, tc.want, JobProxyEnv(tc.envs, tc.cacheURL, tc.services))
+		})
+	}
+}
+
+// docker-in-docker over tcp: the docker client would otherwise send API calls to a proxy
+// that cannot route to the daemon.
+func TestBypassProxyForDockerHost(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		httpProxy  string
+		dockerHost string
+		want       string
+	}{
+		{
+			name:       "tcp daemon is added",
+			httpProxy:  "http://proxy:3128",
+			dockerHost: "tcp://docker:2375",
+			want:       "internal.example,docker",
+		},
+		{
+			name:       "unix socket is left alone",
+			httpProxy:  "http://proxy:3128",
+			dockerHost: "unix:///var/run/docker.sock",
+			want:       "internal.example",
+		},
+		{
+			name:       "nothing happens without a proxy",
+			dockerHost: "tcp://docker:2375",
+			want:       "internal.example",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			clearProxyEnv(t)
+			t.Setenv("no_proxy", "internal.example")
+			if tc.httpProxy != "" {
+				t.Setenv("http_proxy", tc.httpProxy)
+			}
+
+			BypassProxyForDockerHost(tc.dockerHost)
+
+			assert.Equal(t, tc.want, os.Getenv("no_proxy"))
+		})
+	}
+}
+
+func TestProxyPasswords(t *testing.T) {
+	clearProxyEnv(t)
+	t.Setenv("http_proxy", "http://user:hunter2@proxy:3128")
+	t.Setenv("https_proxy", "http://user:s3cret@proxy:3128")
+	assert.Equal(t, []string{"hunter2", "s3cret"}, proxyPasswords())
+
+	t.Setenv("https_proxy", "http://proxy:3128")
+	assert.Equal(t, []string{"hunter2"}, proxyPasswords())
+}
+
+func TestAppendNoProxy(t *testing.T) {
+	assert.Equal(t, "a.example,b.example", appendNoProxy(" a.example , b.example "))
+	// a host already listed is not repeated
+	assert.Equal(t, "cache.local", appendNoProxy("cache.local", "cache.local"))
+	assert.Empty(t, appendNoProxy("", "", ""))
+}
+
+func TestHostOf(t *testing.T) {
+	assert.Equal(t, "cache.local", hostOf("http://cache.local:8088/"))
+	assert.Equal(t, "192.168.1.10", hostOf("http://192.168.1.10:34567"))
+	// nothing to bypass for a socket path or an unparseable URL
+	assert.Empty(t, hostOf("unix:///var/run/docker.sock"))
+	assert.Empty(t, hostOf("cache.local:8088"))
+	assert.Empty(t, hostOf("://nope"))
+}

internal/app/run/runner.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -268,7 +269,8 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
 
 	ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
 	defer cancel()
-	reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg)
+	// A proxy URL may carry credentials, and every job is given it; keep them out of the log.
+	reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg, proxyPasswords()...)
 	var runErr error
 	defer func() {
 		r.runningCount.Add(-1)
@@ -341,6 +343,11 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
 	taskContext := task.Context.Fields
 	envs := r.cloneEnvs()
 
+	// Added per task because this job's service containers must be reached directly, and
+	// act reaches them by their workflow key.
+	proxyEnv := JobProxyEnv(envs, envs["ACTIONS_CACHE_URL"], slices.Sorted(maps.Keys(job.Services)))
+	maps.Copy(envs, proxyEnv)
+
 	if r.capabilities != "" {
 		envs["GITEA_ACTIONS_CAPABILITIES"] = r.capabilities
 	}
@@ -450,6 +457,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
 		LogOutput:            true,
 		JSONLogger:           false,
 		Env:                  envs,
+		ProxyEnv:             proxyEnv,
 		Secrets:              task.Secrets,
 		GitHubInstance:       strings.TrimSuffix(r.client.Address(), "/"),
 		AutoRemove:           true,

internal/app/run/runner_test.go

@@ -93,6 +93,24 @@ func TestNewRunnerInitializesLabelsAndEnvironment(t *testing.T) {
 	require.Nil(t, r.cacheHandler)
 }
 
+// Proxy variables are assembled per task, because a job's service containers have to be
+// reached directly and they are only known once the workflow is parsed.
+func TestNewRunnerLeavesProxyToTheTask(t *testing.T) {
+	clearProxyEnv(t)
+	t.Setenv("http_proxy", "http://proxy:3128")
+
+	cfg := &config.Config{}
+	cfg.Cache.ExternalServer = "http://cache.local:8088/"
+	reg := &config.Registration{Name: "runner"}
+	cli := clientmocks.NewClient(t)
+	cli.On("Address").Return("https://gitea.example/").Maybe()
+
+	r := NewRunner(cfg, reg, cli)
+
+	require.NotContains(t, r.envs, "http_proxy")
+	require.NotContains(t, r.envs, "no_proxy")
+}
+
 func taskWithDefaultActionsURL(url string) *runnerv1.Task {
 	return &runnerv1.Task{
 		Context: &structpb.Struct{

internal/pkg/lock/keyed.go

@@ -0,0 +1,38 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package lock
+
+import "sync"
+
+// Keyed serializes the work done under one key while letting different keys run in
+// parallel. Its zero value is ready to use, and it keeps one mutex per key it has seen.
+type Keyed[K comparable] struct {
+	mu      sync.Mutex
+	mutexes map[K]*sync.Mutex
+}
+
+// Lock locks the mutex for key and returns its unlock function.
+func (kl *Keyed[K]) Lock(key K) func() {
+	kl.mu.Lock()
+	if kl.mutexes == nil {
+		kl.mutexes = map[K]*sync.Mutex{}
+	}
+	lock := kl.mutexes[key]
+	if lock == nil {
+		lock = &sync.Mutex{}
+		kl.mutexes[key] = lock
+	}
+	kl.mu.Unlock()
+
+	lock.Lock()
+	return lock.Unlock
+}
+
+// Delete drops the mutex for key. Holders of it keep working, they just no longer share it
+// with a later Lock of the same key.
+func (kl *Keyed[K]) Delete(key K) {
+	kl.mu.Lock()
+	delete(kl.mutexes, key)
+	kl.mu.Unlock()
+}

internal/pkg/report/reporter.go

@@ -32,6 +32,12 @@ const (
 	maxOutputValueLen = 1024 * 1024 // 1 MiB
 )
 
+// jobOutput is a job output on its way to the server, sent once the server has acknowledged it.
+type jobOutput struct {
+	value string
+	sent  bool
+}
+
 type Reporter struct {
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -53,7 +59,8 @@ type Reporter struct {
 	state             *runnerv1.TaskState
 	stateChanged      bool
 	stateMu           sync.RWMutex
-	outputs           sync.Map
+	outputsMu         sync.Mutex
+	outputs           map[string]jobOutput
 	daemon            chan struct{}
 	heartbeatStop     chan struct{}
 	heartbeatStopOnce sync.Once
@@ -79,8 +86,13 @@ type Reporter struct {
 	stopCommandEndToken string
 }
 
-func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config) *Reporter {
+// extraMasks are values known before the job starts that are not among its secrets, such as
+// the password in the runner's proxy URL.
+func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config, extraMasks ...string) *Reporter {
 	var oldnew []string
+	for _, v := range extraMasks {
+		oldnew = runner.AppendSecretMasker(oldnew, v)
+	}
 	if v := task.Context.Fields["token"].GetStringValue(); v != "" {
 		oldnew = runner.AppendSecretMasker(oldnew, v)
 	}
@@ -394,7 +406,12 @@ func (r *Reporter) logf(format string, a ...any) {
 func (r *Reporter) SetOutputs(outputs map[string]string) {
 	r.stateMu.Lock()
 	defer r.stateMu.Unlock()
+	r.outputsMu.Lock()
+	defer r.outputsMu.Unlock()
 
+	if r.outputs == nil {
+		r.outputs = map[string]jobOutput{}
+	}
 	for k, v := range outputs {
 		if l := len(k); l > maxOutputKeyLen {
 			log.Warnf("ignore output %q because the key is too long: %d > %d", k, l, maxOutputKeyLen)
@@ -406,10 +423,9 @@ func (r *Reporter) SetOutputs(outputs map[string]string) {
 			r.logf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen)
 			continue
 		}
-		if _, ok := r.outputs.Load(k); ok {
-			continue
+		if _, ok := r.outputs[k]; !ok {
+			r.outputs[k] = jobOutput{value: v}
 		}
-		r.outputs.Store(k, v)
 	}
 }
 
@@ -574,14 +590,14 @@ func (r *Reporter) ReportState(reportResult bool) error {
 	r.clientM.Lock()
 	defer r.clientM.Unlock()
 
-	// Build the outputs map first (single Range pass instead of two).
 	outputs := make(map[string]string)
-	r.outputs.Range(func(k, v any) bool {
-		if val, ok := v.(string); ok {
-			outputs[k.(string)] = val
+	r.outputsMu.Lock()
+	for key, out := range r.outputs {
+		if !out.sent {
+			outputs[key] = out.value
 		}
-		return true
-	})
+	}
+	r.outputsMu.Unlock()
 
 	// Consume stateChanged atomically with the snapshot; restored on error
 	// below so a concurrent Fire() during UpdateTask isn't silently lost.
@@ -594,7 +610,8 @@ func (r *Reporter) ReportState(reportResult bool) error {
 		r.stateMu.Unlock()
 		return nil
 	}
-	state := proto.Clone(r.state).(*runnerv1.TaskState)
+	state := &runnerv1.TaskState{}
+	proto.Merge(state, r.state)
 	r.stateChanged = false
 	r.stateMu.Unlock()
 
@@ -622,21 +639,23 @@ func (r *Reporter) ReportState(reportResult bool) error {
 	metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
 	r.lastReportedAtNanos.Store(time.Now().UnixNano())
 
+	var noSent []string
+	r.outputsMu.Lock()
 	for _, k := range resp.Msg.SentOutputs {
-		r.outputs.Store(k, struct{}{})
+		if _, ok := r.outputs[k]; ok {
+			r.outputs[k] = jobOutput{sent: true}
+		}
 	}
+	for key, out := range r.outputs {
+		if !out.sent {
+			noSent = append(noSent, key)
+		}
+	}
+	r.outputsMu.Unlock()
 
 	if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED {
 		r.cancel()
 	}
-
-	var noSent []string
-	r.outputs.Range(func(k, v any) bool {
-		if _, ok := v.(string); ok {
-			noSent = append(noSent, k.(string))
-		}
-		return true
-	})
 	if len(noSent) > 0 {
 		return fmt.Errorf("there are still outputs that have not been sent: %v", noSent)
 	}

internal/pkg/report/reporter_test.go

@@ -7,6 +7,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -1011,33 +1013,59 @@ func TestReporter_SetOutputs(t *testing.T) {
 	r := &Reporter{state: &runnerv1.TaskState{}}
 
 	r.SetOutputs(map[string]string{"foo": "bar"})
-	got, ok := r.outputs.Load("foo")
+	got, ok := r.outputs["foo"]
 	require.True(t, ok)
-	assert.Equal(t, "bar", got)
+	assert.Equal(t, "bar", got.value)
 
 	// first value wins: a later write to the same key is ignored
 	r.SetOutputs(map[string]string{"foo": "baz"})
-	got, _ = r.outputs.Load("foo")
-	assert.Equal(t, "bar", got)
+	assert.Equal(t, "bar", r.outputs["foo"].value)
 
 	// keys longer than maxOutputKeyLen are dropped
 	longKey := strings.Repeat("k", maxOutputKeyLen+1)
 	r.SetOutputs(map[string]string{longKey: "v"})
-	_, ok = r.outputs.Load(longKey)
+	_, ok = r.outputs[longKey]
 	assert.False(t, ok)
 
 	// values longer than maxOutputValueLen are dropped
 	longValue := strings.Repeat("v", maxOutputValueLen+1)
 	r.SetOutputs(map[string]string{"big": longValue})
-	_, ok = r.outputs.Load("big")
+	_, ok = r.outputs["big"]
 	assert.False(t, ok)
 
 	// a value at exactly the limit is still stored
 	maxValue := strings.Repeat("v", maxOutputValueLen)
 	r.SetOutputs(map[string]string{"atlimit": maxValue})
-	got, ok = r.outputs.Load("atlimit")
+	got, ok = r.outputs["atlimit"]
 	require.True(t, ok)
-	assert.Len(t, got, maxOutputValueLen)
+	assert.Len(t, got.value, maxOutputValueLen)
+}
+
+// An output the server acknowledged is not reported again.
+func TestReporter_OutputsSentOnce(t *testing.T) {
+	client := mocks.NewClient(t)
+	var reported []map[string]string
+	client.On("UpdateTask", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
+			reported = append(reported, req.Msg.Outputs)
+			return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Msg.Outputs))}), nil
+		})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	taskCtx, err := structpb.NewStruct(map[string]any{})
+	require.NoError(t, err)
+	cfg, _ := config.LoadDefault("")
+	r := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
+
+	r.SetOutputs(map[string]string{"foo": "bar"})
+	require.NoError(t, r.ReportState(false))
+	assert.True(t, r.outputs["foo"].sent)
+
+	require.NoError(t, r.ReportState(true))
+	require.Len(t, reported, 2)
+	assert.Equal(t, map[string]string{"foo": "bar"}, reported[0])
+	assert.Empty(t, reported[1])
 }
 
 func TestReporter_EffectiveCloseTimeout(t *testing.T) {