refactor(actions): migrate the runner API to connect-go v2

Pin https://gitea.com/gitea/actionslib/pulls/28, which generates
connect-go v2 code. v2 hides plain error messages from clients and caps
messages at 4 MiB. Keep v1's behaviour for runners: they show the messages,
and send all pending log rows in one UpdateLog.

Assisted-by: Claude Code:claude-opus-5-5

go.mod

@@ -5,10 +5,10 @@ go 1.27
 toolchain go1.27.2
 
 require (
-	connectrpc.com/connect/v2 v2.0.0
+	connectrpc.com/connect/v2 v2.0.1
 	gitea.com/lunny/dingtalk_webhook v0.0.0-20171025031554-e3534c89ef96
 	gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4
-	gitea.dev/actionslib v1.4.1
+	gitea.dev/actionslib v1.4.2-0.20261019014634-1fd849368a92
 	gitea.dev/sdk v1.3.0
 	github.com/42wim/httpsig v1.3.0
 	github.com/42wim/sshsig v0.0.0-20260317195500-b9f38cf0d432
@@ -107,7 +107,6 @@ require (
 
 require (
 	cloud.google.com/go/compute/metadata v0.9.0 // indirect
-	connectrpc.com/connect v1.21.0 // indirect
 	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/RoaringBitmap/roaring/v2 v2.18.2 // indirect
 	github.com/STARRY-S/zip v0.2.3 // indirect

go.sum

@@ -2,10 +2,8 @@ cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdB
 cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10=
 code.pfad.fr/check v1.1.0 h1:GWvjdzhSEgHvEHe2uJujDcpmZoySKuHQNrZMfzfO0bE=
 code.pfad.fr/check v1.1.0/go.mod h1:NiUH13DtYsb7xp5wll0U4SXx7KhXQVCtRgdC96IPfoM=
-connectrpc.com/connect v1.21.0 h1:LhqSJt7jHf5NJBo9Jq/t/9FjcYAideif0mg+qe2jCUs=
-connectrpc.com/connect v1.21.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
-connectrpc.com/connect/v2 v2.0.0 h1:PVVpWl2Xv9dIWN2gijs4HK40NSnPFzLqCEEmZ8fCUQ8=
-connectrpc.com/connect/v2 v2.0.0/go.mod h1:N32OLrstcPUzkykFynnefFro3+xzSdOrZlK2jHmS/xo=
+connectrpc.com/connect/v2 v2.0.1 h1:WTAycxe87UqxSM5oxS30d1dalZ+C5xz8Ec8O6zVYkHQ=
+connectrpc.com/connect/v2 v2.0.1/go.mod h1:N32OLrstcPUzkykFynnefFro3+xzSdOrZlK2jHmS/xo=
 filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 gitea.com/lunny/dingtalk_webhook v0.0.0-20171025031554-e3534c89ef96 h1:+wWBi6Qfruqu7xJgjOIrKVQGiLUZdpKYCZewJ4clqhw=
@@ -14,8 +12,8 @@ gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4 h1:IFT+hup2xejHq
 gitea.com/lunny/levelqueue v0.4.2-0.20230414023320-3c0159fe0fe4/go.mod h1:HBqmLbz56JWpfEGG0prskAV97ATNRoj5LDmPicD22hU=
 gitea.com/xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a h1:lSA0F4e9A2NcQSqGqTOXqu2aRi/XEQxDCBwM8yJtE6s=
 gitea.com/xorm/sqlfiddle v0.0.0-20180821085327-62ce714f951a/go.mod h1:EXuID2Zs0pAQhH8yz+DNjUbjppKQzKFAn28TMYPB6IU=
-gitea.dev/actionslib v1.4.1 h1:DWlUnzDfuuMArQWS6yZ+vwJu9av7sZVMOox7li/kyu4=
-gitea.dev/actionslib v1.4.1/go.mod h1:RbXaZudZ04YndjBBxZJF9rZAqU6SKGkaSD1sob5lQbE=
+gitea.dev/actionslib v1.4.2-0.20261019014634-1fd849368a92 h1:qjj0mqRCwZ269pJo4LH8e4ePMVK/iWMA0tO6Dxiey+M=
+gitea.dev/actionslib v1.4.2-0.20261019014634-1fd849368a92/go.mod h1:NWzFSnEG5EEXaq66Ql3moZ31g4H7+sAmSCyBirjj2R4=
 gitea.dev/sdk v1.3.0 h1:4U95+Ou6WHCQiRPDEyAzvl1Q7cxg5RUTWkqmOXr5w4I=
 gitea.dev/sdk v1.3.0/go.mod h1:85EBqSAqUn1KlFCk40y3sTjWjBsB/KouL3slGXMA/2o=
 github.com/42wim/httpsig v1.3.0 h1:HK2bieeZ99X9IvOoc1ST6yrhLek0f4MqlEGES2JCP3g=

routers/api/actions/ping/ping.go

@@ -13,10 +13,15 @@ import (
 	"gitea.dev/modules/log"
 
 	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 )
 
 func NewPingServiceHandler() (string, http.Handler) {
-	return pingv1connect.NewPingServiceHandler(&Service{})
+	server := connect.NewServer()
+	pingv1connect.RegisterPingServiceHandler(server, &Service{})
+	mux := http.NewServeMux()
+	connecthttp.Mount(mux, server)
+	return "/" + pingv1connect.PingServiceName + "/", mux
 }
 
 var _ pingv1connect.PingServiceHandler = (*Service)(nil)
@@ -25,12 +30,13 @@ type Service struct{}
 
 func (s *Service) Ping(
 	ctx context.Context,
-	req *connect.Request[pingv1.PingRequest],
-) (*connect.Response[pingv1.PingResponse], error) {
-	log.Trace("Content-Type: %s", req.Header().Get("Content-Type"))
-	log.Trace("User-Agent: %s", req.Header().Get("User-Agent"))
-	res := connect.NewResponse(&pingv1.PingResponse{
-		Data: fmt.Sprintf("Hello, %s!", req.Msg.Data),
-	})
+	req *pingv1.PingRequest,
+) (*pingv1.PingResponse, error) {
+	info, _ := connect.CallInfoForServerContext(ctx)
+	log.Trace("Content-Type: %s", info.RequestHeader().Get("Content-Type"))
+	log.Trace("User-Agent: %s", info.RequestHeader().Get("User-Agent"))
+	res := &pingv1.PingResponse{
+		Data: fmt.Sprintf("Hello, %s!", req.Data),
+	}
 	return res, nil
 }

routers/api/actions/ping/ping_test.go

@@ -12,16 +12,14 @@ import (
 	"gitea.dev/actionslib/ping/v1/pingv1connect"
 
 	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestService(t *testing.T) {
-	mux := http.NewServeMux()
-	mux.Handle(pingv1connect.NewPingServiceHandler(
-		&Service{},
-	))
-	MainServiceTest(t, mux)
+	_, handler := NewPingServiceHandler()
+	MainServiceTest(t, handler)
 }
 
 func MainServiceTest(t *testing.T, h http.Handler) {
@@ -31,31 +29,31 @@ func MainServiceTest(t *testing.T, h http.Handler) {
 	server.StartTLS()
 	defer server.Close()
 
-	connectClient := pingv1connect.NewPingServiceClient(
+	connectClient := pingv1connect.NewPingServiceClient(connect.NewClient(connecthttp.NewTransport(
 		server.Client(),
 		server.URL,
-	)
+	)))
 
-	grpcClient := pingv1connect.NewPingServiceClient(
+	grpcClient := pingv1connect.NewPingServiceClient(connect.NewClient(connecthttp.NewTransport(
 		server.Client(),
 		server.URL,
-		connect.WithGRPC(),
-	)
+		connecthttp.WithGRPC(),
+	)))
 
-	grpcWebClient := pingv1connect.NewPingServiceClient(
+	grpcWebClient := pingv1connect.NewPingServiceClient(connect.NewClient(connecthttp.NewTransport(
 		server.Client(),
 		server.URL,
-		connect.WithGRPCWeb(),
-	)
+		connecthttp.WithGRPCWeb(),
+	)))
 
 	clients := []pingv1connect.PingServiceClient{connectClient, grpcClient, grpcWebClient}
 	t.Run("ping request", func(t *testing.T) {
 		for _, client := range clients {
-			result, err := client.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{
+			result, err := client.Ping(t.Context(), &pingv1.PingRequest{
 				Data: "foobar",
-			}))
+			})
 			require.NoError(t, err)
-			assert.Equal(t, "Hello, foobar!", result.Msg.Data)
+			assert.Equal(t, "Hello, foobar!", result.Data)
 		}
 	})
 }

routers/api/actions/runner/interceptor.go

@@ -26,27 +26,39 @@ const (
 	tokenHeaderKey = protocol.TokenHeader
 )
 
-var withRunner = connect.WithInterceptors(connect.UnaryInterceptorFunc(func(unaryFunc connect.UnaryFunc) connect.UnaryFunc {
+// withErrorMessages sends plain errors to runners as CodeUnknown with their message, like connect v1 did.
+func withErrorMessages(next connect.ServerFunc) connect.ServerFunc {
+	return func(ctx context.Context, spec connect.Spec, stream connect.ServerStream) error {
+		err := next(ctx, spec, stream)
+		if _, ok := errors.AsType[*connect.Error](err); err == nil || ok {
+			return err
+		}
+		return connect.NewError(connect.CodeUnknown, err.Error()).WithCause(err)
+	}
+}
+
+func withRunner(next connect.ServerFunc) connect.ServerFunc {
 	// A plain gRPC status.Error is treated as unknown by Connect and becomes HTTP 500
 	// To respond an HTTP status code, use connect.Error instead
-	errUnregisteredRunner := connect.NewError(connect.CodeUnauthenticated, errors.New("unregistered runner"))
-	return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-		methodName := getMethodName(request)
+	errUnregisteredRunner := connect.NewError(connect.CodeUnauthenticated, "unregistered runner")
+	return func(ctx context.Context, spec connect.Spec, stream connect.ServerStream) error {
+		methodName := getMethodName(spec)
 		if methodName == "Register" {
-			return unaryFunc(ctx, request)
+			return next(ctx, spec, stream)
 		}
-		uuid := request.Header().Get(uuidHeaderKey)
-		token := request.Header().Get(tokenHeaderKey)
+		info, _ := connect.CallInfoForServerContext(ctx)
+		uuid := info.RequestHeader().Get(uuidHeaderKey)
+		token := info.RequestHeader().Get(tokenHeaderKey)
 
 		runner, err := actions_model.GetRunnerByUUID(ctx, uuid)
 		if err != nil {
 			if errors.Is(err, util.ErrNotExist) {
-				return nil, errUnregisteredRunner
+				return errUnregisteredRunner
 			}
-			return nil, status.Error(codes.Internal, err.Error())
+			return status.Error(codes.Internal, err.Error())
 		}
 		if !util.CryptoConstTimeEqual(runner.TokenHash, auth_model.HashToken(token, runner.TokenSalt)) {
-			return nil, errUnregisteredRunner
+			return errUnregisteredRunner
 		}
 
 		now := time.Now()
@@ -72,12 +84,12 @@ var withRunner = connect.WithInterceptors(connect.UnaryInterceptorFunc(func(unar
 		}
 
 		ctx = context.WithValue(ctx, runnerCtxKey{}, runner)
-		return unaryFunc(ctx, request)
+		return next(ctx, spec, stream)
 	}
-}))
+}
 
-func getMethodName(req connect.AnyRequest) string {
-	splits := strings.Split(req.Spec().Procedure, "/")
+func getMethodName(spec connect.Spec) string {
+	splits := strings.Split(spec.Procedure, "/")
 	if len(splits) > 0 {
 		return splits[len(splits)-1]
 	}

routers/api/actions/runner/runner.go

@@ -21,17 +21,19 @@ import (
 	actions_service "gitea.dev/services/actions"
 
 	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
 func NewRunnerServiceHandler() (string, http.Handler) {
-	return runnerv1connect.NewRunnerServiceHandler(
-		&Service{},
-		connect.WithCompressMinBytes(1024),
-		withRunner,
-	)
+	server := connect.NewServer(withErrorMessages, withRunner)
+	runnerv1connect.RegisterRunnerServiceHandler(server, &Service{})
+	mux := http.NewServeMux()
+	// runners send all pending log rows in one UpdateLog, so keep v1's unlimited message size
+	connecthttp.Mount(mux, server, connecthttp.WithCompressMinBytes(1024), connecthttp.WithReadMaxBytes(0))
+	return "/" + runnerv1connect.RunnerServiceName + "/", mux
 }
 
 var _ runnerv1connect.RunnerServiceClient = (*Service)(nil)
@@ -41,13 +43,13 @@ type Service struct{}
 // Register for new runner.
 func (s *Service) Register(
 	ctx context.Context,
-	req *connect.Request[runnerv1.RegisterRequest],
-) (*connect.Response[runnerv1.RegisterResponse], error) {
-	if req.Msg.Token == "" || req.Msg.Name == "" {
+	req *runnerv1.RegisterRequest,
+) (*runnerv1.RegisterResponse, error) {
+	if req.Token == "" || req.Name == "" {
 		return nil, errors.New("missing runner token, name")
 	}
 
-	runnerToken, err := actions_model.GetRunnerToken(ctx, req.Msg.Token)
+	runnerToken, err := actions_model.GetRunnerToken(ctx, req.Token)
 	if err != nil {
 		return nil, errors.New("runner registration token not found")
 	}
@@ -68,19 +70,19 @@ func (s *Service) Register(
 		}
 	}
 
-	labels := req.Msg.Labels
-	hasCancellingSupport := slices.Contains(req.Msg.GetCapabilities(), runnerCapabilityCancelling)
+	labels := req.Labels
+	hasCancellingSupport := slices.Contains(req.GetCapabilities(), runnerCapabilityCancelling)
 
 	// create new runner
-	name := util.EllipsisDisplayString(req.Msg.Name, 255)
+	name := util.EllipsisDisplayString(req.Name, 255)
 	runner := &actions_model.ActionRunner{
 		UUID:                 uuid.New().String(),
 		Name:                 name,
 		OwnerID:              runnerToken.OwnerID,
 		RepoID:               runnerToken.RepoID,
-		Version:              req.Msg.Version,
+		Version:              req.Version,
 		AgentLabels:          labels,
-		Ephemeral:            req.Msg.Ephemeral,
+		Ephemeral:            req.Ephemeral,
 		HasCancellingSupport: hasCancellingSupport,
 	}
 	runner.GenerateAndFillToken()
@@ -96,7 +98,7 @@ func (s *Service) Register(
 		return nil, errors.New("can't update runner token status")
 	}
 
-	res := connect.NewResponse(&runnerv1.RegisterResponse{
+	res := &runnerv1.RegisterResponse{
 		Runner: &runnerv1.Runner{
 			Id:        runner.ID,
 			Uuid:      runner.UUID,
@@ -106,7 +108,7 @@ func (s *Service) Register(
 			Labels:    runner.AgentLabels,
 			Ephemeral: runner.Ephemeral,
 		},
-	})
+	}
 
 	return res, nil
 }
@@ -139,14 +141,18 @@ func applyDeclareRequestToRunner(runner *actions_model.ActionRunner, req declare
 
 func (s *Service) Declare(
 	ctx context.Context,
-	req *connect.Request[runnerv1.DeclareRequest],
-) (*connect.Response[runnerv1.DeclareResponse], error) {
+	req *runnerv1.DeclareRequest,
+) (*runnerv1.DeclareResponse, error) {
 	runner := GetRunner(ctx)
-	if err := actions_model.UpdateRunner(ctx, runner, applyDeclareRequestToRunner(runner, req.Msg)...); err != nil {
+	if err := actions_model.UpdateRunner(ctx, runner, applyDeclareRequestToRunner(runner, req)...); err != nil {
 		return nil, status.Errorf(codes.Internal, "update runner: %v", err)
 	}
 
-	resp := connect.NewResponse(&runnerv1.DeclareResponse{
+	// Capabilities are communicated via headers to avoid a hard dependency on a proto bump.
+	// Older runners ignore unknown headers; newer runners can use this for feature negotiation.
+	info, _ := connect.CallInfoForServerContext(ctx)
+	info.ResponseHeader().Set("X-Gitea-Actions-Capabilities", actions_model.RunnerCapabilities())
+	return &runnerv1.DeclareResponse{
 		Runner: &runnerv1.Runner{
 			Id:      runner.ID,
 			Uuid:    runner.UUID,
@@ -155,22 +161,18 @@ func (s *Service) Declare(
 			Version: runner.Version,
 			Labels:  runner.AgentLabels,
 		},
-	})
-	// Capabilities are communicated via headers to avoid a hard dependency on a proto bump.
-	// Older runners ignore unknown headers; newer runners can use this for feature negotiation.
-	resp.Header().Set("X-Gitea-Actions-Capabilities", actions_model.RunnerCapabilities())
-	return resp, nil
+	}, nil
 }
 
 // FetchTask assigns a task to the runner
 func (s *Service) FetchTask(
 	ctx context.Context,
-	req *connect.Request[runnerv1.FetchTaskRequest],
-) (*connect.Response[runnerv1.FetchTaskResponse], error) {
+	req *runnerv1.FetchTaskRequest,
+) (*runnerv1.FetchTaskResponse, error) {
 	runner := GetRunner(ctx)
 
 	var task *runnerv1.Task
-	tasksVersion := req.Msg.TasksVersion // task version from runner
+	tasksVersion := req.TasksVersion // task version from runner
 	latestVersion, err := actions_model.GetTasksVersionByScope(ctx, runner.OwnerID, runner.RepoID)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "query tasks version failed: %v", err)
@@ -207,26 +209,26 @@ func (s *Service) FetchTask(
 			task = t
 		}
 	}
-	res := connect.NewResponse(&runnerv1.FetchTaskResponse{
+	res := &runnerv1.FetchTaskResponse{
 		Task:         task,
 		TasksVersion: latestVersion,
-	})
+	}
 	return res, nil
 }
 
 // UpdateTask updates the task status.
 func (s *Service) UpdateTask(
 	ctx context.Context,
-	req *connect.Request[runnerv1.UpdateTaskRequest],
-) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
+	req *runnerv1.UpdateTaskRequest,
+) (*runnerv1.UpdateTaskResponse, error) {
 	runner := GetRunner(ctx)
 
-	task, err := actions_model.UpdateTaskByState(ctx, runner.ID, req.Msg.State)
+	task, err := actions_model.UpdateTaskByState(ctx, runner.ID, req.State)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "update task: %v", err)
 	}
 
-	for k, v := range req.Msg.Outputs {
+	for k, v := range req.Outputs {
 		if len(k) > 255 {
 			log.Warn("Ignore the output of task %d because the key is too long: %q", task.ID, k)
 			continue
@@ -264,7 +266,7 @@ func (s *Service) UpdateTask(
 		actions_service.NotifyWorkflowJobStatusUpdateWithTask(ctx, task.Job, task)
 	}
 
-	if req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED {
+	if req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED {
 		if err := actions_service.EmitJobsIfReadyByRun(task.Job.RunID); err != nil {
 			log.Error("Emit ready jobs of run %d: %v", task.Job.RunID, err)
 		}
@@ -273,25 +275,25 @@ func (s *Service) UpdateTask(
 		}
 	}
 
-	return connect.NewResponse(&runnerv1.UpdateTaskResponse{
+	return &runnerv1.UpdateTaskResponse{
 		State: &runnerv1.TaskState{
-			Id:     req.Msg.State.Id,
+			Id:     req.State.Id,
 			Result: task.Status.AsResult(),
 		},
 		SentOutputs: sentOutputs,
-	}), nil
+	}, nil
 }
 
 // UpdateLog uploads log of the task.
 func (s *Service) UpdateLog(
 	ctx context.Context,
-	req *connect.Request[runnerv1.UpdateLogRequest],
-) (*connect.Response[runnerv1.UpdateLogResponse], error) {
+	req *runnerv1.UpdateLogRequest,
+) (*runnerv1.UpdateLogResponse, error) {
 	runner := GetRunner(ctx)
 
-	res := connect.NewResponse(&runnerv1.UpdateLogResponse{})
+	res := &runnerv1.UpdateLogResponse{}
 
-	task, err := actions_model.GetTaskByID(ctx, req.Msg.TaskId)
+	task, err := actions_model.GetTaskByID(ctx, req.TaskId)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "get task: %v", err)
 	} else if runner.ID != task.RunnerID {
@@ -301,8 +303,8 @@ func (s *Service) UpdateLog(
 
 	// Trim rows the runner already had acked.
 	var rows []*runnerv1.LogRow
-	if req.Msg.Index <= ack && int64(len(req.Msg.Rows))+req.Msg.Index > ack {
-		rows = req.Msg.Rows[ack-req.Msg.Index:]
+	if req.Index <= ack && int64(len(req.Rows))+req.Index > ack {
+		rows = req.Rows[ack-req.Index:]
 	}
 
 	// Ack a re-sent finalize idempotently. Appending new rows past the seal errors.
@@ -310,15 +312,15 @@ func (s *Service) UpdateLog(
 		if len(rows) > 0 {
 			return nil, status.Errorf(codes.AlreadyExists, "log file has been archived")
 		}
-		res.Msg.AckIndex = ack
+		res.AckIndex = ack
 		return res, nil
 	}
 
 	// Bail unless we have new rows or a NoMore to finalize. Even with
 	// NoMore, bail when the runner has outrun the server — archiving a
 	// log with a gap is worse than asking it to retry.
-	if len(rows) == 0 && (!req.Msg.NoMore || req.Msg.Index > ack) {
-		res.Msg.AckIndex = ack
+	if len(rows) == 0 && (!req.NoMore || req.Index > ack) {
+		res.AckIndex = ack
 		return res, nil
 	}
 
@@ -335,10 +337,10 @@ func (s *Service) UpdateLog(
 		task.LogSize += int64(n)
 	}
 
-	res.Msg.AckIndex = task.LogLength
+	res.AckIndex = task.LogLength
 
 	var remove func()
-	if req.Msg.NoMore {
+	if req.NoMore {
 		task.LogInStorage = true
 		remove, err = actions.TransferLogs(ctx, task.LogFilename)
 		if err != nil {

routers/api/actions/runner/runner_test.go

@@ -4,14 +4,33 @@
 package runner
 
 import (
+	"errors"
+	"net/http/httptest"
 	"testing"
 
 	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
 	actions_model "gitea.dev/models/actions"
 
+	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+func TestPlainErrorReachesRunnerAsUnknownWithMessage(t *testing.T) {
+	_, handler := NewRunnerServiceHandler()
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	client := runnerv1connect.NewRunnerServiceClient(connect.NewClient(connecthttp.NewTransport(server.Client(), server.URL)))
+	_, err := client.Register(t.Context(), &runnerv1.RegisterRequest{})
+	connectErr, ok := errors.AsType[*connect.Error](err)
+	require.True(t, ok)
+	assert.Equal(t, connect.CodeUnknown, connectErr.Code())
+	assert.Equal(t, "missing runner token, name", connectErr.Message())
+}
+
 func TestApplyDeclareRequestToRunnerAdvertisedCapabilityEnablesCancelling(t *testing.T) {
 	runner := &actions_model.ActionRunner{}
 	req := &runnerv1.DeclareRequest{

tests/integration/actions_job_test.go

@@ -27,7 +27,6 @@ import (
 	"gitea.dev/modules/timeutil"
 	actions_service "gitea.dev/services/actions"
 
-	"connectrpc.com/connect/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -643,33 +642,33 @@ jobs:
 		err = actions_service.CleanupEphemeralRunners(t.Context())
 		assert.NoError(t, err)
 
-		resp, err := runner.client.runnerServiceClient.FetchTask(t.Context(), connect.NewRequest(&runnerv1.FetchTaskRequest{
+		resp, err := runner.client.runnerServiceClient.FetchTask(t.Context(), &runnerv1.FetchTaskRequest{
 			TasksVersion: 0,
-		}))
+		})
 		assert.NoError(t, err)
-		assert.Nil(t, resp.Msg.Task)
+		assert.Nil(t, resp.Task)
 
 		// verify CleanupEphemeralRunners does not remove this runner
 		err = actions_service.CleanupEphemeralRunners(t.Context())
 		assert.NoError(t, err)
 
-		_, err = runner.client.runnerServiceClient.UpdateTask(t.Context(), connect.NewRequest(&runnerv1.UpdateTaskRequest{
+		_, err = runner.client.runnerServiceClient.UpdateTask(t.Context(), &runnerv1.UpdateTaskRequest{
 			State: &runnerv1.TaskState{
 				Id:     actionTask.ID,
 				Result: runnerv1.Result_RESULT_SUCCESS,
 			},
-		}))
+		})
 		assert.NoError(t, err)
 
-		resp, err = runner.client.runnerServiceClient.FetchTask(t.Context(), connect.NewRequest(&runnerv1.FetchTaskRequest{
+		resp, err = runner.client.runnerServiceClient.FetchTask(t.Context(), &runnerv1.FetchTaskRequest{
 			TasksVersion: 0,
-		}))
+		})
 		assert.Error(t, err)
 		assert.Nil(t, resp)
 
-		resp, err = runner.client.runnerServiceClient.FetchTask(t.Context(), connect.NewRequest(&runnerv1.FetchTaskRequest{
+		resp, err = runner.client.runnerServiceClient.FetchTask(t.Context(), &runnerv1.FetchTaskRequest{
 			TasksVersion: 0,
-		}))
+		})
 		assert.Error(t, err)
 		assert.Nil(t, resp)
 

tests/integration/actions_log_finalize_test.go

@@ -19,7 +19,6 @@ import (
 	actions_module "gitea.dev/modules/actions"
 	"gitea.dev/modules/storage"
 
-	"connectrpc.com/connect/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -56,14 +55,14 @@ jobs:
 
 		task := runner.fetchTask(t)
 
-		resp, err := runner.client.runnerServiceClient.UpdateLog(t.Context(), connect.NewRequest(&runnerv1.UpdateLogRequest{
+		resp, err := runner.client.runnerServiceClient.UpdateLog(t.Context(), &runnerv1.UpdateLogRequest{
 			TaskId: task.Id,
 			Index:  0,
 			Rows:   nil,
 			NoMore: true,
-		}))
+		})
 		require.NoError(t, err)
-		assert.EqualValues(t, 0, resp.Msg.AckIndex)
+		assert.EqualValues(t, 0, resp.AckIndex)
 
 		freshTask := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionTask{ID: task.Id})
 		require.True(t, freshTask.LogInStorage, "log_in_storage must flip after empty NoMore=true")
@@ -78,13 +77,13 @@ jobs:
 		// A sealed log must ack the re-send and still reject new appended rows.
 		t.Run("re-sent finalize is idempotent", func(t *testing.T) {
 			finalize := &runnerv1.UpdateLogRequest{TaskId: task.Id, Index: 0, Rows: nil, NoMore: true}
-			resp, err := runner.client.runnerServiceClient.UpdateLog(t.Context(), connect.NewRequest(finalize))
+			resp, err := runner.client.runnerServiceClient.UpdateLog(t.Context(), finalize)
 			require.NoError(t, err)
-			assert.EqualValues(t, 0, resp.Msg.AckIndex)
+			assert.EqualValues(t, 0, resp.AckIndex)
 
-			_, err = runner.client.runnerServiceClient.UpdateLog(t.Context(), connect.NewRequest(&runnerv1.UpdateLogRequest{
+			_, err = runner.client.runnerServiceClient.UpdateLog(t.Context(), &runnerv1.UpdateLogRequest{
 				TaskId: task.Id, Index: 0, Rows: []*runnerv1.LogRow{{Content: "late"}}, NoMore: true,
-			}))
+			})
 			require.Error(t, err, "appending rows past the seal must be rejected")
 		})
 	})

tests/integration/actions_runner_test.go

@@ -18,6 +18,7 @@ import (
 	"gitea.dev/modules/setting"
 
 	"connectrpc.com/connect/v2"
+	"connectrpc.com/connect/v2/connecthttp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -40,45 +41,44 @@ func newMockRunner() *mockRunner {
 func newMockRunnerClient(uuid, token string) *mockRunnerClient {
 	baseURL := setting.AppURL + "api/actions"
 
-	opt := connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
-		return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+	client := connect.NewClient(connecthttp.NewTransport(http.DefaultClient, baseURL), func(next connect.ClientFunc) connect.ClientFunc {
+		return func(ctx context.Context, spec connect.Spec) (connect.ClientStream, error) {
+			info, _ := connect.CallInfoForClientContext(ctx)
 			if uuid != "" {
-				req.Header().Set("x-runner-uuid", uuid)
+				info.RequestHeader().Set("x-runner-uuid", uuid)
 			}
 			if token != "" {
-				req.Header().Set("x-runner-token", token)
+				info.RequestHeader().Set("x-runner-token", token)
 			}
-			return next(ctx, req)
+			return next(ctx, spec)
 		}
-	}))
+	})
 
-	client := &mockRunnerClient{
-		pingServiceClient:   pingv1connect.NewPingServiceClient(http.DefaultClient, baseURL, opt),
-		runnerServiceClient: runnerv1connect.NewRunnerServiceClient(http.DefaultClient, baseURL, opt),
+	return &mockRunnerClient{
+		pingServiceClient:   pingv1connect.NewPingServiceClient(client),
+		runnerServiceClient: runnerv1connect.NewRunnerServiceClient(client),
 	}
-
-	return client
 }
 
 func (r *mockRunner) doPing(t *testing.T) {
-	resp, err := r.client.pingServiceClient.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{
+	resp, err := r.client.pingServiceClient.Ping(t.Context(), &pingv1.PingRequest{
 		Data: "mock-runner",
-	}))
+	})
 	assert.NoError(t, err)
-	assert.Equal(t, "Hello, mock-runner!", resp.Msg.Data)
+	assert.Equal(t, "Hello, mock-runner!", resp.Data)
 }
 
 func (r *mockRunner) doRegister(t *testing.T, name, token string, labels []string, ephemeral bool) {
 	r.doPing(t)
-	resp, err := r.client.runnerServiceClient.Register(t.Context(), connect.NewRequest(&runnerv1.RegisterRequest{
+	resp, err := r.client.runnerServiceClient.Register(t.Context(), &runnerv1.RegisterRequest{
 		Name:      name,
 		Token:     token,
 		Version:   "mock-runner-version",
 		Labels:    labels,
 		Ephemeral: ephemeral,
-	}))
+	})
 	assert.NoError(t, err)
-	r.client = newMockRunnerClient(resp.Msg.Runner.Uuid, resp.Msg.Runner.Token)
+	r.client = newMockRunnerClient(resp.Runner.Uuid, resp.Runner.Token)
 }
 
 func (r *mockRunner) registerAsRepoRunner(t *testing.T, ownerName, repoName, runnerName string, labels []string, ephemeral bool) {
@@ -127,11 +127,11 @@ func (r *mockRunner) tryFetchTask(t *testing.T, timeout ...time.Duration) *runne
 // and returns the task (if any) and the TasksVersion from the response.
 // Used to verify the production path where the runner sends the current version.
 func (r *mockRunner) fetchTaskOnce(t *testing.T, tasksVersion int64) (*runnerv1.Task, int64) {
-	resp, err := r.client.runnerServiceClient.FetchTask(t.Context(), connect.NewRequest(&runnerv1.FetchTaskRequest{
+	resp, err := r.client.runnerServiceClient.FetchTask(t.Context(), &runnerv1.FetchTaskRequest{
 		TasksVersion: tasksVersion,
-	}))
+	})
 	require.NoError(t, err)
-	return resp.Msg.Task, resp.Msg.TasksVersion
+	return resp.Task, resp.TasksVersion
 }
 
 type mockTaskOutcome struct {
@@ -142,35 +142,35 @@ type mockTaskOutcome struct {
 
 func (r *mockRunner) execTask(t *testing.T, task *runnerv1.Task, outcome *mockTaskOutcome) {
 	for idx, lr := range outcome.logRows {
-		resp, err := r.client.runnerServiceClient.UpdateLog(t.Context(), connect.NewRequest(&runnerv1.UpdateLogRequest{
+		resp, err := r.client.runnerServiceClient.UpdateLog(t.Context(), &runnerv1.UpdateLogRequest{
 			TaskId: task.Id,
 			Index:  int64(idx),
 			Rows:   []*runnerv1.LogRow{lr},
 			NoMore: idx == len(outcome.logRows)-1,
-		}))
+		})
 		assert.NoError(t, err)
-		assert.EqualValues(t, idx+1, resp.Msg.AckIndex)
+		assert.EqualValues(t, idx+1, resp.AckIndex)
 	}
 	sentOutputKeys := make([]string, 0, len(outcome.outputs))
 	for outputKey, outputValue := range outcome.outputs {
-		resp, err := r.client.runnerServiceClient.UpdateTask(t.Context(), connect.NewRequest(&runnerv1.UpdateTaskRequest{
+		resp, err := r.client.runnerServiceClient.UpdateTask(t.Context(), &runnerv1.UpdateTaskRequest{
 			State: &runnerv1.TaskState{
 				Id:     task.Id,
 				Result: runnerv1.Result_RESULT_UNSPECIFIED,
 			},
 			Outputs: map[string]string{outputKey: outputValue},
-		}))
+		})
 		assert.NoError(t, err)
 		sentOutputKeys = append(sentOutputKeys, outputKey)
-		assert.ElementsMatch(t, sentOutputKeys, resp.Msg.SentOutputs)
+		assert.ElementsMatch(t, sentOutputKeys, resp.SentOutputs)
 	}
-	resp, err := r.client.runnerServiceClient.UpdateTask(t.Context(), connect.NewRequest(&runnerv1.UpdateTaskRequest{
+	resp, err := r.client.runnerServiceClient.UpdateTask(t.Context(), &runnerv1.UpdateTaskRequest{
 		State: &runnerv1.TaskState{
 			Id:        task.Id,
 			Result:    outcome.result,
 			StoppedAt: timestamppb.Now(),
 		},
-	}))
+	})
 	assert.NoError(t, err)
-	assert.Equal(t, outcome.result, resp.Msg.State.Result)
+	assert.Equal(t, outcome.result, resp.State.Result)
 }