Adjust RunnerJobRequest support

actionsdotnetactcompat/act_runner.go

@@ -10,10 +10,8 @@ type ActRunner struct {
 }
 
 func (arunner *ActRunner) ExecWorker(run *actionsrunner.RunRunner, wc actionsrunner.WorkerContext, jobreq *protocol.AgentJobRequestMessage, src []byte) error {
-	jlogger := wc.Logger()
-	jobExecCtx := wc.JobExecCtx()
 	if len(arunner.WorkerArgs) <= 0 {
-		ExecWorker(jobreq, jlogger, jobExecCtx)
+		ExecWorker(jobreq, wc)
 		return nil
 	}
 	return arunner.WorkerRunnerEnvironment.ExecWorker(run, wc, jobreq, src)

actionsdotnetactcompat/act_worker.go

@@ -2,14 +2,14 @@ package actionsdotnetactcompat
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"fmt"
 	"os"
 	"regexp"
 	"runtime"
 	"strings"
 
+	"github.com/ChristopherHX/github-act-runner/actionsrunner"
 	"github.com/ChristopherHX/github-act-runner/protocol"
 	"github.com/google/uuid"
 	"github.com/nektos/act/pkg/common"
@@ -100,7 +100,9 @@ func (f *ghaFormatter) Format(entry *logrus.Entry) ([]byte, error) {
 	return b.Bytes(), nil
 }
 
-func ExecWorker(rqt *protocol.AgentJobRequestMessage, jlogger *protocol.JobLogger, jobExecCtx context.Context) {
+func ExecWorker(rqt *protocol.AgentJobRequestMessage, wc actionsrunner.WorkerContext) {
+	jlogger := wc.Logger()
+	jobExecCtx := wc.JobExecCtx()
 	logger := logrus.New()
 	logger.SetOutput(jlogger)
 	formatter := &ghaFormatter{
@@ -114,14 +116,7 @@ func ExecWorker(rqt *protocol.AgentJobRequestMessage, jlogger *protocol.JobLogge
 		jlogger.TimelineRecords.Value[0].Complete(result)
 		jlogger.Logger.Close()
 		jlogger.Finish()
-		finish := &protocol.JobEvent{
-			Name:      "JobCompleted",
-			JobID:     rqt.JobID,
-			RequestID: rqt.RequestID,
-			Result:    result,
-			Outputs:   outputs,
-		}
-		vssConnection.FinishJob(finish, rqt.Plan)
+		wc.FinishJob(result, outputs)
 	}
 	finishJob := func(result string) {
 		finishJob2(result, &map[string]protocol.VariableValue{})

actionsrunner/runner.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"path"
 	"runtime"
 	"runtime/debug"
 	"strings"
@@ -36,6 +37,7 @@ type JobRun struct {
 	Plan            *protocol.TaskOrchestrationPlanReference
 	Name            string
 	RegistrationURL string
+	RunServiceURL   string
 }
 
 type RunnerEnvironment interface {
@@ -439,6 +441,7 @@ func runJob(runnerenv RunnerEnvironment, joblock *sync.Mutex, vssConnection *pro
 			plogger.Printf("%v\n", string(src))
 		}
 		jobreq := &protocol.AgentJobRequestMessage{}
+		var runServiceUrl string
 		{
 			if strings.EqualFold(message.MessageType, "RunnerJobRequest") {
 				rjrr := &RunnerJobRequestRef{}
@@ -452,13 +455,14 @@ func runJob(runnerenv RunnerEnvironment, joblock *sync.Mutex, vssConnection *pro
 					} else {
 						copy := *vssConnection
 						vssConnection = &copy
-						vssConnection.Token = ""
-						runServiceUrl, _ := url.ParseRequestURI(rjrr.RunServiceUrl)
-						vssConnection.TenantURL = runServiceUrl.Host
+						runServiceUrl = rjrr.RunServiceUrl
+						acquirejobUrl, _ := url.Parse(runServiceUrl)
+						acquirejobUrl.Path = path.Join(acquirejobUrl.Path, "acquirejob")
+						vssConnection.TenantURL = runServiceUrl
 						payload := &RunnerServicePayload{
 							StreamID: rjrr.RunnerRequestId,
 						}
-						err = vssConnection.RequestWithContext2(jobctx, "POST", rjrr.RunServiceUrl, "", payload, &src)
+						err = vssConnection.RequestWithContext2(jobctx, "POST", acquirejobUrl.String(), "", payload, &src)
 					}
 					if err == nil {
 						json.Unmarshal(src, jobreq)
@@ -476,6 +480,7 @@ func runJob(runnerenv RunnerEnvironment, joblock *sync.Mutex, vssConnection *pro
 			Plan:            jobreq.Plan,
 			RegistrationURL: instance.RegistrationURL,
 			Name:            instance.Agent.Name,
+			RunServiceURL:   runServiceUrl,
 		}
 		{
 			// TODO multi repository runners can receive multiple job requests at the same time and this protection doesn't work there

actionsrunner/worker_context.go

@@ -2,7 +2,10 @@ package actionsrunner
 
 import (
 	"context"
+	"path"
+	"strings"
 	"time"
+	"net/url"
 
 	"github.com/ChristopherHX/github-act-runner/protocol"
 )
@@ -24,6 +27,37 @@ type DefaultWorkerContext struct {
 }
 
 func (wc *DefaultWorkerContext) FinishJob(result string, outputs *map[string]protocol.VariableValue) {
+	if strings.EqualFold(wc.Message().MessageType, "RunnerJobRequest") {
+		payload := struct {
+			PlanID     string
+			JobID      string
+			Conclusion string
+			Outputs    *map[string]protocol.VariableValue
+		}{
+			PlanID: wc.Message().Plan.PlanID,
+    		JobID: wc.Message().JobID,
+    		Conclusion: result,
+    		Outputs: outputs,
+		}
+
+		completejobUrl, _ := url.Parse(wc.VssConnection.TenantURL)
+		completejobUrl.Path = path.Join(completejobUrl.Path, "completejob")
+		for i := 0; ; i++ {
+			if err := wc.VssConnection.RequestWithContext2(context.Background(), "POST", completejobUrl.String(), "", payload, nil); err != nil {
+				wc.RunnerLogger.Printf("Failed to finish Job '%v' with Status %v: %v\n", wc.Message().JobDisplayName, result, err.Error())
+			} else {
+				wc.RunnerLogger.Printf("Finished Job '%v' with Status %v\n", wc.Message().JobDisplayName, result)
+				break
+			}
+			if i < 10 {
+				wc.RunnerLogger.Printf("Retry finishing '%v' in 10 seconds attempt %v of 10\n", wc.Message().JobDisplayName, i+1)
+				<-time.After(time.Second * 10)
+			} else {
+				break
+			}
+		}
+		return
+	}
 	finish := &protocol.JobEvent{
 		Name:      "JobCompleted",
 		JobID:     wc.Message().JobID,

main.go

@@ -303,7 +303,7 @@ func main() {
 							wc.Init()
 							wc.Logger().Append(protocol.CreateTimelineEntry(jobreq.JobID, "__setup", "Set up Job")).Start()
 							wc.Logger().Update()
-							actionsdotnetactcompat.ExecWorker(jobreq, wc.Logger(), execcontext)
+							actionsdotnetactcompat.ExecWorker(jobreq, wc)
 							logf.WriteString("Finish Job\n")
 						}()
 					default: