Clean up live ephemeral runners after terminal jobs

apis/actions.github.com/v1alpha1/ephemeralrunner_types.go

@@ -183,6 +183,21 @@ type EphemeralRunnerStatus struct {
 
 	// +optional
 	JobDisplayName string `json:"jobDisplayName,omitempty"`
+
+	// JobCompletion records the terminal job event received for this runner.
+	// +optional
+	JobCompletion *EphemeralRunnerJobCompletion `json:"jobCompletion,omitempty"`
+}
+
+// EphemeralRunnerJobCompletion identifies the terminal job event received for
+// an ephemeral runner. The controller uses all identity fields before deleting
+// a runner whose process did not exit after the job became terminal.
+type EphemeralRunnerJobCompletion struct {
+	Result        string      `json:"result"`
+	RunnerID      int         `json:"runnerId"`
+	JobID         string      `json:"jobId"`
+	WorkflowRunID int64       `json:"workflowRunId"`
+	FinishedAt    metav1.Time `json:"finishedAt"`
 }
 
 // EphemeralRunnerPhase is the phase of the ephemeral runner.

cmd/ghalistener/scaler/scaler.go

@@ -12,6 +12,7 @@ import (
 	"github.com/actions/scaleset/listener"
 	jsonpatch "github.com/evanphx/json-patch"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -156,6 +157,59 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar
 
 func (w *Scaler) HandleJobCompleted(ctx context.Context, msg *scaleset.JobCompleted) error {
 	w.dirty = true
+
+	w.logger.Info("Recording completed job for the runner",
+		"runnerName", msg.RunnerName,
+		"runnerId", msg.RunnerID,
+		"jobId", msg.JobID,
+		"workflowRunId", msg.WorkflowRunID,
+		"result", msg.Result,
+		"finishedAt", msg.FinishTime)
+
+	original, err := json.Marshal(&v1alpha1.EphemeralRunner{})
+	if err != nil {
+		return fmt.Errorf("failed to marshal empty ephemeral runner: %w", err)
+	}
+
+	patch, err := json.Marshal(&v1alpha1.EphemeralRunner{
+		Status: v1alpha1.EphemeralRunnerStatus{
+			JobCompletion: &v1alpha1.EphemeralRunnerJobCompletion{
+				Result:        msg.Result,
+				RunnerID:      msg.RunnerID,
+				JobID:         msg.JobID,
+				WorkflowRunID: msg.WorkflowRunID,
+				FinishedAt:    metav1.NewTime(msg.FinishTime),
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal ephemeral runner completion patch: %w", err)
+	}
+
+	mergePatch, err := jsonpatch.CreateMergePatch(original, patch)
+	if err != nil {
+		return fmt.Errorf("failed to create completion merge patch for ephemeral runner: %w", err)
+	}
+
+	patchedStatus := &v1alpha1.EphemeralRunner{}
+	err = w.clientset.RESTClient().
+		Patch(types.MergePatchType).
+		Prefix("apis", v1alpha1.GroupVersion.Group, v1alpha1.GroupVersion.Version).
+		Namespace(w.config.EphemeralRunnerSetNamespace).
+		Resource("EphemeralRunners").
+		Name(msg.RunnerName).
+		SubResource("status").
+		Body(mergePatch).
+		Do(ctx).
+		Into(patchedStatus)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			w.logger.Info("Ephemeral runner not found, skipping completed job status patch", "runnerName", msg.RunnerName)
+			return nil
+		}
+		return fmt.Errorf("could not patch completed job status, patch JSON: %s, error: %w", string(mergePatch), err)
+	}
+
 	return nil
 }
 

cmd/ghalistener/scaler/scaler_test.go

@@ -1,15 +1,97 @@
 package scaler
 
 import (
+	"context"
+	"encoding/json"
+	"io"
 	"log/slog"
 	"math"
+	"net/http"
+	"net/http/httptest"
 	"testing"
+	"time"
 
+	"github.com/actions/scaleset"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 )
 
 var discardLogger = slog.New(slog.DiscardHandler)
 
+func TestHandleJobCompleted_RecordsTerminalStatusForExactRunner(t *testing.T) {
+	t.Parallel()
+
+	type recordedRequest struct {
+		method string
+		path   string
+		body   []byte
+	}
+
+	requests := make(chan recordedRequest, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		requests <- recordedRequest{method: r.Method, path: r.URL.Path, body: body}
+		w.Header().Set("Content-Type", "application/json")
+		_, err = w.Write([]byte(`{"apiVersion":"actions.github.com/v1alpha1","kind":"EphemeralRunner"}`))
+		require.NoError(t, err)
+	}))
+	t.Cleanup(server.Close)
+
+	clientset, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
+	require.NoError(t, err)
+
+	worker := &Scaler{
+		clientset: clientset,
+		config: Config{
+			EphemeralRunnerSetNamespace: "arc-runners",
+			EphemeralRunnerSetName:      "linux-x64",
+		},
+		logger: discardLogger,
+	}
+	finishedAt := time.Date(2026, time.August, 14, 0, 30, 37, 0, time.UTC)
+	completion := &scaleset.JobCompleted{
+		Result:     "failed",
+		RunnerID:   2402,
+		RunnerName: "linux-x64-runner-ctwvm",
+		JobMessageBase: scaleset.JobMessageBase{
+			JobID:         "85cb98eb-2919-5766-872c-cc997f618c1f",
+			WorkflowRunID: 31753891150,
+			FinishTime:    finishedAt,
+		},
+	}
+
+	require.NoError(t, worker.HandleJobCompleted(context.Background(), completion))
+
+	select {
+	case request := <-requests:
+		assert.Equal(t, http.MethodPatch, request.method)
+		assert.Equal(t, "/apis/actions.github.com/v1alpha1/namespaces/arc-runners/ephemeralrunners/linux-x64-runner-ctwvm/status", request.path)
+
+		var patch struct {
+			Status struct {
+				JobCompletion struct {
+					Result        string    `json:"result"`
+					RunnerID      int       `json:"runnerId"`
+					JobID         string    `json:"jobId"`
+					WorkflowRunID int64     `json:"workflowRunId"`
+					FinishedAt    time.Time `json:"finishedAt"`
+				} `json:"jobCompletion"`
+			} `json:"status"`
+		}
+		require.NoError(t, json.Unmarshal(request.body, &patch))
+		assert.Equal(t, completion.Result, patch.Status.JobCompletion.Result)
+		assert.Equal(t, completion.RunnerID, patch.Status.JobCompletion.RunnerID)
+		assert.Equal(t, completion.JobID, patch.Status.JobCompletion.JobID)
+		assert.Equal(t, completion.WorkflowRunID, patch.Status.JobCompletion.WorkflowRunID)
+		assert.Equal(t, completion.FinishTime, patch.Status.JobCompletion.FinishedAt)
+	case <-time.After(time.Second):
+		t.Fatal("JobCompleted was acknowledged without recording terminal state on the exact EphemeralRunner")
+	}
+}
+
 func TestSetDesiredWorkerState_MinMaxDefaults(t *testing.T) {
 	newEmptyWorker := func() *Scaler {
 		return &Scaler{

controllers/actions.github.com/ephemeralrunner_controller.go

@@ -44,6 +44,7 @@ import (
 const (
 	ephemeralRunnerFinalizerName        = "ephemeralrunner.actions.github.com/finalizer"
 	ephemeralRunnerActionsFinalizerName = "ephemeralrunner.actions.github.com/runner-registration-finalizer"
+	completedJobRunnerGracePeriod       = 30 * time.Second
 )
 
 // EphemeralRunnerReconciler reconciles a EphemeralRunner object
@@ -186,6 +187,29 @@ func (r *EphemeralRunnerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		log.Info("Successfully added finalizers")
 	}
 
+	if completion := ephemeralRunner.Status.JobCompletion; completion != nil {
+		if !jobCompletionMatchesRunner(&ephemeralRunner) {
+			log.Info("Ignoring completed job event that does not match the ephemeral runner status",
+				"completionRunnerId", completion.RunnerID,
+				"completionJobId", completion.JobID,
+				"completionWorkflowRunId", completion.WorkflowRunID)
+		} else {
+			deleteAfter := completion.FinishedAt.Add(completedJobRunnerGracePeriod)
+			if wait := time.Until(deleteAfter); wait > 0 {
+				log.Info("Waiting briefly for the runner process to exit after job completion", "requeueAfter", wait)
+				return ctrl.Result{RequeueAfter: wait}, nil
+			}
+
+			log.Info("Job is terminal but runner is still present; issuing delete",
+				"result", completion.Result,
+				"finishedAt", completion.FinishedAt)
+			if err := r.Delete(ctx, &ephemeralRunner); client.IgnoreNotFound(err) != nil {
+				return ctrl.Result{}, fmt.Errorf("failed to delete ephemeral runner after terminal job: %w", err)
+			}
+			return ctrl.Result{}, nil
+		}
+	}
+
 	secret := new(corev1.Secret)
 	if err := r.Get(ctx, req.NamespacedName, secret); err != nil {
 		if !kerrors.IsNotFound(err) {
@@ -971,6 +995,18 @@ func runnerContainerStatus(pod *corev1.Pod) *corev1.ContainerStatus {
 	return nil
 }
 
+func jobCompletionMatchesRunner(runner *v1alpha1.EphemeralRunner) bool {
+	completion := runner.Status.JobCompletion
+	if completion == nil || completion.Result == "" || completion.JobID == "" || completion.FinishedAt.IsZero() {
+		return false
+	}
+
+	return completion.RunnerID == runner.Status.RunnerID &&
+		runner.Status.RunnerName == runner.Name &&
+		completion.JobID == runner.Status.JobID &&
+		completion.WorkflowRunID == runner.Status.WorkflowRunID
+}
+
 func initContainerFailed(pod *corev1.Pod) bool {
 	for i := range pod.Status.InitContainerStatuses {
 		cs := &pod.Status.InitContainerStatuses[i]