fix: fix panics, enable the forcetypeassert lint (#1123)

An unchecked type assertion panics on input it did not expect, as a missing `tool_cache` key did in https://gitea.com/gitea/runner/pulls/1122.

Every flagged site is now handled where it can fail, or typed so it cannot: a `lock.Keyed` replaces the two `sync.Map` mutex registries, and the reporter's outputs carry an explicit sent flag. Mocks keep their assertions, a mismatch there is a setup error the panic names.

Bugs it turned up (only the first is reachable from workflows):

1. A scalar `matrix.include` or `matrix.exclude`, e.g. `include: foo` or `include: [1, 2]`, panicked the runner with `interface conversion: interface {} is string, not map[string]interface {}`. Verified against `main`, it is now a workflow error. `OnSchedule` panicked the same way on a malformed `on.schedule` entry.
1. `ExternalURL()` panicked on the nil listener after `Close()`, the port is now resolved once at startup.
1. `errors.Is(err, git.ErrShortRef)` followed by `err.(*git.Error)` panics as soon as anything wraps that error, so it is `errors.As` now.
1. An output name the server acknowledged without ever being sent one was recorded as sent forever, which silently dropped a later value for that name.

48576ab3e5 fixes one discovered issue: a matrix key holding a nested object was logged and then run as if the job had no matrix, so it now fails like an unknown `exclude` key.
Reviewed-on: https://gitea.com/gitea/runner/pulls/1123
Reviewed-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: silverwind <me@silverwind.io>

.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

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",
+	return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
-		h.outboundIP,
-		h.listener.Addr().(*net.TCPAddr).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{})
+	return cloneLocks.Lock(dir)
-	mu := v.(*sync.Mutex)
-	mu.Lock()
-	return mu.Unlock
 }
 
 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/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) {
+			entry, ok := v.(map[string]any)
-				if k != "cron" {
+			if !ok {
-					continue
+				continue
-				}
+			}
-				allSchedules = append(allSchedules, cron.(string))
+			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, err
-			return nil
 		}
 		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)
+						if err := addInclude(i); err != nil {
-						extraInclude := true
+							return nil, err
-						for k := range i {
-							if _, ok := m[k]; ok {
-								includes = append(includes, i)
-								extraInclude = false
-								break
-							}
-						}
-						if extraInclude {
-							extraIncludes = append(extraIncludes, i)
 						}
 					}
 				case any:
-					v := v.(map[string]any)
+					if err := addInclude(t); err != nil {
-					extraInclude := true
+						return nil, err
-					for k := range v {
-						if _, ok := m[k]; ok {
-							includes = append(includes, v)
-							extraInclude = false
-							break
-						}
-					}
-					if extraInclude {
-						extraIncludes = append(extraIncludes, v)
 					}
 				}
 			}
@@ -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)
+				exclude, ok := e.(map[string]any)
-				for k := range e {
+				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, err := job.Matrix()
-		matrix := job.Matrix()
+		require.ErrorContains(t, err, `matrix key "config" has invalid nested object value`)
-
+		assert.Nil(t, matrix)
-		// Should return nil due to validation error
-		assert.Nil(t, matrix, "matrix with nested map should return nil")
 	})
 }
 

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 := cachedActionTar(ctx, step, stepModel.Uses, "")
-		ta, err := rc.Config.ActionCache.GetTarArchive(ctx, raction.cacheDir, raction.resolvedSha, "")
 		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 = cachedActionTar(ctx, step, actionName, contextDir)
-				buildContext, err = rc.Config.ActionCache.GetTarArchive(ctx, rstep.cacheDir, rstep.resolvedSha, contextDir)
 				if err != nil {
 					return err
 				}

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"
@@ -715,13 +715,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{})
+	return jobMutexes.Lock(job)
-	mu := v.(*sync.Mutex)
-	mu.Lock()
-	return mu.Unlock
 }
 
 func (rc *RunContext) interpolateOutputs() common.Executor {

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 {

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
@@ -394,7 +401,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 +418,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 {
+		if _, ok := r.outputs[k]; !ok {
-			continue
+			r.outputs[k] = jobOutput{value: v}
 		}
-		r.outputs.Store(k, v)
 	}
 }
 
@@ -574,14 +585,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 {
+	r.outputsMu.Lock()
-		if val, ok := v.(string); ok {
+	for key, out := range r.outputs {
-			outputs[k.(string)] = val
+		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 +605,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 +634,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", r.outputs["foo"].value)
-	assert.Equal(t, "bar", got)
 
 	// 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) {