fix: scope the action identity to the step executing an action (#1359)

`github.action_repository`/`github.action_ref` and their env vars now belong to the step executing an action, like on GitHub: set for remote actions, empty for `run`, `docker://` and local steps. Before, a local action nested in a remote composite resolved `$/` against the remote action instead of the workflow's repository.

Updates actionslib to v1.5.1, which accepts `$/` as the repository root action.

Co-authored-by: bircni <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1359
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>

DEVELOPMENT.md

@@ -59,5 +59,10 @@ Each lives in `internal/pkg/action/<name>` and is registered in `builtinActions`
 It requires Docker and is excluded from `make test`. CI runs stable and nightly variants in
 parallel.
 
+The `cli_*` scenarios build the binary and drive `register` and `daemon` as a user would, including `--once`,
+`--labels` and shutdown with a running job. `failure_paths` covers failed, skipped, tolerated and timed-out jobs.
+`label_routing` and `capacity` cover scheduling, `reconnect` drops the connection to Gitea through a proxy, and
+`host_mode` and `reusable_workflow` cover host labels and `workflow_call`.
+
 The suite shares one Gitea and regular runner. Cache and ephemeral scenarios use isolated
 repository runners. A run path is `<workflow>@<ref>` and a log row is `<timestamp>Z <payload>`.

act/artifacts/server.go

@@ -218,6 +218,7 @@ func Serve(ctx context.Context, artifactPath, addr, port string) context.CancelF
 	logger.Debugf("Artifacts base path '%s'", artifactPath)
 	uploads(router, artifactPath)
 	downloads(router, artifactPath)
+	registerV4(router, artifactPath)
 
 	server := &http.Server{
 		Addr:              net.JoinHostPort(addr, port),

act/artifacts/server_test.go

@@ -28,6 +28,7 @@ func TestArtifactFlow(t *testing.T) {
 	router := http.NewServeMux()
 	uploads(router, artifactPath)
 	downloads(router, artifactPath)
+	registerV4(router, artifactPath)
 	server := httptest.NewServer(router)
 	defer server.Close()
 
@@ -174,6 +175,92 @@ func TestArtifactFlow(t *testing.T) {
 		require.Equal(t, http.StatusOK, status)
 		require.Equal(t, content, string(data))
 	})
+
+	t.Run("v4", func(t *testing.T) {
+		const artifactRef = `"workflow_run_backend_id":"7","name":"50% art@v1"`
+		call := func(method, body string) (int, map[string]any) {
+			status, data := request(t, http.MethodPost, baseURL+artifactServicePath+"/"+method, strings.NewReader(body), nil)
+			var out map[string]any
+			require.NoError(t, json.Unmarshal(data, &out), string(data))
+			return status, out
+		}
+		put := func(rawURL, body string) int {
+			status, _ := request(t, http.MethodPut, rawURL, strings.NewReader(body), nil)
+			return status
+		}
+		download := func() (string, http.Header) {
+			status, signed := call("GetSignedArtifactURL", "{"+artifactRef+"}")
+			require.Equal(t, http.StatusOK, status, signed)
+			resp, err := client.Get(signed["signedUrl"].(string))
+			require.NoError(t, err)
+			defer resp.Body.Close()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+			body, err := io.ReadAll(resp.Body)
+			require.NoError(t, err)
+			return string(body), resp.Header
+		}
+
+		status, created := call("CreateArtifact", "{"+artifactRef+"}")
+		require.Equal(t, http.StatusOK, status, created)
+		upload := created["signedUploadUrl"].(string)
+		require.Equal(t, http.StatusCreated, put(upload+"&comp=block&blockid=Yg", "world"))
+		require.Equal(t, http.StatusCreated, put(upload+"&comp=block&blockid=YQ", "hello "))
+		for attempt := range 2 {
+			require.Equal(t, http.StatusCreated, put(upload+"&comp=blocklist", `<?xml version="1.0"?><BlockList><Latest>YQ</Latest><Latest>Yg</Latest></BlockList>`), "commit attempt %d", attempt)
+		}
+
+		digest := "sha256:b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"
+		status, _ = call("FinalizeArtifact", "{"+artifactRef+`,"size":"10"}`)
+		require.Equal(t, http.StatusBadRequest, status)
+		status, finalized := call("FinalizeArtifact", "{"+artifactRef+`,"size":"11","hash":"`+digest+`"}`)
+		require.Equal(t, http.StatusOK, status, finalized)
+		id := finalized["artifactId"].(string)
+		status, refinalized := call("FinalizeArtifact", "{"+artifactRef+`,"size":"10"}`)
+		require.Equal(t, http.StatusOK, status)
+		require.Equal(t, id, refinalized["artifactId"])
+		status, conflict := call("CreateArtifact", "{"+artifactRef+"}")
+		require.Equal(t, http.StatusConflict, status)
+		require.Equal(t, "already_exists", conflict["code"])
+
+		status, listed := call("ListArtifacts", `{"workflowRunBackendId":"7","idFilter":"`+id+`"}`)
+		require.Equal(t, http.StatusOK, status)
+		artifacts := listed["artifacts"].([]any)
+		require.Len(t, artifacts, 1)
+		artifact := artifacts[0].(map[string]any)
+		require.Regexp(t, `^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$`, artifact["createdAt"])
+		delete(artifact, "createdAt")
+		require.Equal(t, map[string]any{"workflowRunBackendId": "7", "name": "50% art@v1", "size": "11", "digest": digest, "databaseId": id}, artifact)
+		_, other := call("ListArtifacts", `{"workflow_run_backend_id":"8"}`)
+		require.Empty(t, other["artifacts"])
+
+		body, header := download()
+		require.Equal(t, "hello world", body)
+		require.Equal(t, "application/zip", header.Get("Content-Type"))
+		require.Equal(t, "attachment; filename*=UTF-8''50%25%20art%40v1.zip", header.Get("Content-Disposition"))
+
+		status, _ = call("DeleteArtifact", "{"+artifactRef+"}")
+		require.Equal(t, http.StatusOK, status)
+		status, missing := call("GetSignedArtifactURL", "{"+artifactRef+"}")
+		require.Equal(t, http.StatusNotFound, status)
+		require.Equal(t, "not_found", missing["code"])
+
+		status, created = call("CreateArtifact", "{"+artifactRef+"}")
+		require.Equal(t, http.StatusOK, status, created)
+		status, _ = request(t, http.MethodPut, created["signedUploadUrl"].(string), strings.NewReader("gz"), http.Header{"X-Ms-Blob-Content-Type": {"application/gzip"}})
+		require.Equal(t, http.StatusCreated, status)
+		status, _ = call("FinalizeArtifact", "{"+artifactRef+`,"size":"2"}`)
+		require.Equal(t, http.StatusOK, status)
+		body, header = download()
+		require.Equal(t, "gz", body)
+		require.Equal(t, "application/gzip", header.Get("Content-Type"))
+		require.Equal(t, "attachment; filename*=UTF-8''50%25%20art%40v1", header.Get("Content-Disposition"))
+
+		rerun := http.NewServeMux()
+		registerV4(rerun, artifactPath)
+		recorder := httptest.NewRecorder()
+		rerun.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, artifactServicePath+"/CreateArtifact", strings.NewReader("{"+artifactRef+"}")))
+		require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
+	})
 }
 
 func TestSafeResolve(t *testing.T) {

act/artifacts/server_v4.go

@@ -0,0 +1,329 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package artifacts
+
+import (
+	"cmp"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json/v2"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+)
+
+type serverV4 struct { // the subset of Gitea's v4 artifact service that the toolkit uses, unauthenticated like the v3 routes beside it
+	dir     string
+	mu      sync.Mutex
+	lastID  int64
+	created map[string]bool
+}
+
+const (
+	zipMimeType   = "application/zip"
+	maxTwirpBytes = 1 << 20
+)
+
+func registerV4(router *http.ServeMux, baseDir string) {
+	server := &serverV4{dir: filepath.Join(baseDir, "v4"), lastID: time.Now().UnixMilli(), created: map[string]bool{}} // IDs stay unique across invocations sharing baseDir
+	router.HandleFunc("POST "+artifactServicePath+"/CreateArtifact", twirp(server.createArtifact))
+	router.HandleFunc("POST "+artifactServicePath+"/FinalizeArtifact", twirp(server.finalizeArtifact))
+	router.HandleFunc("POST "+artifactServicePath+"/ListArtifacts", twirp(server.listArtifacts))
+	router.HandleFunc("POST "+artifactServicePath+"/GetSignedArtifactURL", twirp(server.getSignedArtifactURL))
+	router.HandleFunc("POST "+artifactServicePath+"/DeleteArtifact", twirp(server.deleteArtifact))
+	router.HandleFunc("PUT "+artifactServicePath+"/UploadArtifact", server.uploadArtifact)
+	router.HandleFunc("GET "+artifactServicePath+"/DownloadArtifact", server.downloadArtifact)
+}
+
+type twirpError struct {
+	status int
+	code   string
+	msg    string
+}
+
+func (e *twirpError) Error() string { return e.msg }
+
+func notFound(name string) error {
+	return &twirpError{http.StatusNotFound, "not_found", fmt.Sprintf("artifact %q not found", name)}
+}
+
+func badRequest(format string, args ...any) error {
+	return &twirpError{http.StatusBadRequest, "invalid_argument", fmt.Sprintf(format, args...)}
+}
+
+func twirp[Req, Resp any](handle func(*http.Request, Req) (Resp, error)) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var req Req
+		var resp Resp
+		err := json.UnmarshalRead(io.LimitReader(r.Body, maxTwirpBytes), &req, json.MatchCaseInsensitiveNames(true)) // also matches the toolkit's proto field names
+		if err != nil {
+			err = badRequest("malformed request: %v", err)
+		} else {
+			resp, err = handle(r, req)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		if err != nil {
+			terr := &twirpError{http.StatusInternalServerError, "internal", err.Error()}
+			errors.As(err, &terr)
+			w.WriteHeader(terr.status)
+			_ = json.MarshalWrite(w, map[string]string{"code": terr.code, "msg": terr.msg})
+			return
+		}
+		_ = json.MarshalWrite(w, resp)
+	}
+}
+
+func fileName(name string) string { // safe on any filesystem for any client-chosen name, case-insensitive ones included
+	sum := sha256.Sum256([]byte(name))
+	return hex.EncodeToString(sum[:])
+}
+
+func (s *serverV4) runDir(runID string) string {
+	return filepath.Join(s.dir, fileName(runID))
+}
+
+func (s *serverV4) artifactDir(runID, name string) string {
+	return filepath.Join(s.runDir(runID), fileName(name))
+}
+
+func (s *serverV4) urlDir(r *http.Request) string {
+	return s.artifactDir(r.URL.Query().Get("run"), r.URL.Query().Get("name"))
+}
+
+func readMeta(dir string) (*monolithArtifact, error) {
+	data, err := os.ReadFile(filepath.Join(dir, "meta.json"))
+	if err != nil {
+		return nil, err
+	}
+	var meta monolithArtifact
+	return &meta, json.Unmarshal(data, &meta)
+}
+
+func writeMeta(dir string, meta *monolithArtifact) error {
+	data, err := json.Marshal(meta)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(dir, "meta.json"), data, 0o644)
+}
+
+func blobURL(r *http.Request, endpoint, runID, name string) string {
+	return "http://" + r.Host + artifactServicePath + "/" + endpoint + "?" + url.Values{"run": {runID}, "name": {name}}.Encode()
+}
+
+func (s *serverV4) createArtifact(r *http.Request, req createArtifactRequest) (createArtifactResponse, error) { // an exec invocation re-runs its workflows, so it replaces earlier invocations' artifacts but conflicts with its own
+	dir := s.artifactDir(req.WorkflowRunBackendID, req.Name)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.created[dir] {
+		return createArtifactResponse{}, &twirpError{http.StatusConflict, "already_exists", "an artifact with this name already exists on the workflow run"}
+	}
+	if err := os.RemoveAll(dir); err != nil {
+		return createArtifactResponse{}, err
+	}
+	if err := os.MkdirAll(filepath.Join(dir, "blocks"), 0o755); err != nil {
+		return createArtifactResponse{}, err
+	}
+	s.lastID++
+	if err := writeMeta(dir, &monolithArtifact{
+		WorkflowRunBackendID: req.WorkflowRunBackendID,
+		DatabaseID:           s.lastID,
+		Name:                 req.Name,
+		CreatedAt:            time.Now().UTC().Truncate(time.Second),
+	}); err != nil {
+		return createArtifactResponse{}, err
+	}
+	s.created[dir] = true
+	return createArtifactResponse{Ok: true, SignedUploadURL: blobURL(r, "UploadArtifact", req.WorkflowRunBackendID, req.Name)}, nil
+}
+
+type blockList struct {
+	Blocks []string `xml:",any"` // Latest, Committed and Uncommitted all name a staged block
+}
+
+func (s *serverV4) uploadArtifact(w http.ResponseWriter, r *http.Request) {
+	dir := s.urlDir(r)
+	if meta, err := readMeta(dir); err != nil || meta.Digest != "" {
+		http.Error(w, "no artifact awaits an upload", http.StatusNotFound)
+		return
+	}
+	contentType := r.Header.Get("x-ms-blob-content-type")
+	var err error
+	switch comp := r.URL.Query().Get("comp"); comp {
+	case "block":
+		err = writeFile(filepath.Join(dir, "blocks", fileName(r.URL.Query().Get("blockid"))), r.Body)
+	case "blocklist":
+		err = commitBlob(dir, contentType, func(out io.Writer) error { return joinBlocks(out, dir, r.Body) })
+	case "": // Put Blob, which docker/build-push-action uses
+		err = commitBlob(dir, contentType, func(out io.Writer) error {
+			_, err := io.Copy(out, r.Body)
+			return err
+		})
+	default:
+		http.Error(w, "unsupported comp "+comp, http.StatusBadRequest)
+		return
+	}
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusCreated)
+}
+
+func writeFile(name string, body io.Reader) error {
+	file, err := os.Create(name)
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(file, body)
+	return errors.Join(err, file.Close())
+}
+
+func commitBlob(dir, contentType string, write func(io.Writer) error) error { // replaces the blob whole and keeps its content type, as Azure does
+	out, err := os.CreateTemp(dir, "blob")
+	if err != nil {
+		return err
+	}
+	if err = errors.Join(write(out), out.Close()); err == nil {
+		err = os.Rename(out.Name(), filepath.Join(dir, "blob"))
+	}
+	if err != nil {
+		return errors.Join(err, os.Remove(out.Name()))
+	}
+	return os.WriteFile(filepath.Join(dir, "content-type"), []byte(contentType), 0o644)
+}
+
+func joinBlocks(out io.Writer, dir string, body io.Reader) error { // keeps the blocks for finalizeArtifact, so that a retried commit can join them again
+	var list blockList
+	if err := xml.NewDecoder(io.LimitReader(body, maxTwirpBytes)).Decode(&list); err != nil {
+		return err
+	}
+	for _, id := range list.Blocks {
+		if err := appendBlock(out, filepath.Join(dir, "blocks", fileName(id))); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func appendBlock(out io.Writer, name string) error {
+	in, err := os.Open(name)
+	if err != nil {
+		return err
+	}
+	defer in.Close()
+	_, err = io.Copy(out, in)
+	return err
+}
+
+func (s *serverV4) finalizeArtifact(_ *http.Request, req finalizeArtifactRequest) (finalizeArtifactResponse, error) {
+	dir := s.artifactDir(req.WorkflowRunBackendID, req.Name)
+	meta, err := readMeta(dir)
+	if err != nil {
+		return finalizeArtifactResponse{}, notFound(req.Name)
+	} else if meta.Digest != "" {
+		return finalizeArtifactResponse{Ok: true, ArtifactID: meta.DatabaseID}, nil
+	}
+	blob, err := os.Open(filepath.Join(dir, "blob"))
+	if err != nil {
+		return finalizeArtifactResponse{}, badRequest("artifact %q has no upload", req.Name)
+	}
+	defer blob.Close()
+	hash := sha256.New()
+	size, err := io.Copy(hash, blob)
+	if err != nil {
+		return finalizeArtifactResponse{}, err
+	}
+	digest := "sha256:" + hex.EncodeToString(hash.Sum(nil))
+	if size != req.Size {
+		return finalizeArtifactResponse{}, badRequest("size %d does not match the %d bytes uploaded", req.Size, size)
+	}
+	if req.Hash != "" && req.Hash != digest {
+		return finalizeArtifactResponse{}, badRequest("hash %s does not match the uploaded %s", req.Hash, digest)
+	}
+	s.mu.Lock() // only after hashing, so finalizing one large artifact does not stall the others
+	defer s.mu.Unlock()
+	if current, err := readMeta(dir); err != nil || current.DatabaseID != meta.DatabaseID {
+		return finalizeArtifactResponse{}, notFound(req.Name)
+	}
+	meta.Size, meta.Digest = size, digest
+	if err := writeMeta(dir, meta); err != nil {
+		return finalizeArtifactResponse{}, err
+	}
+	return finalizeArtifactResponse{Ok: true, ArtifactID: meta.DatabaseID}, os.RemoveAll(filepath.Join(dir, "blocks"))
+}
+
+func (s *serverV4) listArtifacts(_ *http.Request, req listArtifactsRequest) (listArtifactsResponse, error) {
+	run := s.runDir(req.WorkflowRunBackendID)
+	entries, err := os.ReadDir(run)
+	if err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return listArtifactsResponse{}, err
+	}
+	var resp listArtifactsResponse
+	for _, entry := range entries {
+		meta, err := readMeta(filepath.Join(run, entry.Name()))
+		if err != nil || meta.Digest == "" ||
+			req.NameFilter != "" && req.NameFilter != meta.Name || req.IDFilter != 0 && req.IDFilter != meta.DatabaseID {
+			continue
+		}
+		resp.Artifacts = append(resp.Artifacts, *meta)
+	}
+	slices.SortFunc(resp.Artifacts, func(left, right monolithArtifact) int { return cmp.Compare(left.DatabaseID, right.DatabaseID) })
+	return resp, nil
+}
+
+func (s *serverV4) getSignedArtifactURL(r *http.Request, req getSignedArtifactURLRequest) (getSignedArtifactURLResponse, error) {
+	if meta, err := readMeta(s.artifactDir(req.WorkflowRunBackendID, req.Name)); err != nil || meta.Digest == "" {
+		return getSignedArtifactURLResponse{}, notFound(req.Name)
+	}
+	return getSignedArtifactURLResponse{SignedURL: blobURL(r, "DownloadArtifact", req.WorkflowRunBackendID, req.Name)}, nil
+}
+
+func (s *serverV4) downloadArtifact(w http.ResponseWriter, r *http.Request) {
+	dir := s.urlDir(r)
+	meta, err := readMeta(dir)
+	if err != nil || meta.Digest == "" {
+		http.Error(w, "artifact not found", http.StatusNotFound)
+		return
+	}
+	blob, err := os.Open(filepath.Join(dir, "blob"))
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	defer blob.Close()
+	data, _ := os.ReadFile(filepath.Join(dir, "content-type"))
+	contentType, filename := string(data), meta.Name
+	if !strings.Contains(contentType, "/") {
+		contentType = zipMimeType
+	}
+	if contentType == zipMimeType && !strings.HasSuffix(filename, ".zip") {
+		filename += ".zip"
+	}
+	w.Header().Set("Content-Type", contentType)
+	w.Header().Set("Content-Disposition", "attachment; filename*=UTF-8''"+strings.ReplaceAll(url.QueryEscape(filename), "+", "%20")) // RFC 5987 attr-chars only, which the toolkit and mime.ParseMediaType both accept
+	http.ServeContent(w, r, "", meta.CreatedAt, blob)
+}
+
+func (s *serverV4) deleteArtifact(_ *http.Request, req deleteArtifactRequest) (deleteArtifactResponse, error) {
+	dir := s.artifactDir(req.WorkflowRunBackendID, req.Name)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	meta, err := readMeta(dir)
+	if err != nil {
+		return deleteArtifactResponse{}, notFound(req.Name)
+	}
+	delete(s.created, dir)
+	return deleteArtifactResponse{Ok: true, ArtifactID: meta.DatabaseID}, os.RemoveAll(dir)
+}

act/artifacts/token.go

@@ -0,0 +1,23 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package artifacts
+
+import (
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json/v2"
+	"time"
+)
+
+func LocalRuntimeToken(run string, lifetime time.Duration) string { // an unsecured JWT with the claims the toolkit and buildx read, its run hashed for scp and its jti an unguessable bearer
+	now := time.Now()
+	claims, _ := json.Marshal(map[string]any{
+		"scp": "Actions.Results:" + fileName(run) + ":1",
+		"ac":  `[{"Scope":"","Permission":3}]`,
+		"nbf": now.Unix(),
+		"exp": now.Add(lifetime).Unix(),
+		"jti": rand.Text(),
+	})
+	return base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"none","typ":"JWT"}`)) + "." + base64.RawURLEncoding.EncodeToString(claims) + "."
+}

act/artifacts/twirp.go

@@ -0,0 +1,55 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package artifacts
+
+import "time"
+
+const artifactServicePath = "/twirp/github.actions.results.api.v1.ArtifactService"
+
+type ( // the ArtifactService messages in the protobuf JSON mapping, which writes int64 as strings
+	createArtifactResponse struct {
+		Ok              bool   `json:"ok"`
+		SignedUploadURL string `json:"signedUploadUrl"`
+	}
+
+	finalizeArtifactRequest struct {
+		WorkflowRunBackendID string
+		Name                 string
+		Size                 int64 `json:",string"`
+		Hash                 string
+	}
+	finalizeArtifactResponse struct {
+		Ok         bool  `json:"ok"`
+		ArtifactID int64 `json:"artifactId,string"`
+	}
+
+	listArtifactsRequest struct {
+		WorkflowRunBackendID string
+		NameFilter           string
+		IDFilter             int64 `json:",string"`
+	}
+	listArtifactsResponse struct {
+		Artifacts []monolithArtifact `json:"artifacts"`
+	}
+	monolithArtifact struct {
+		WorkflowRunBackendID string    `json:"workflowRunBackendId"` // the toolkit sends it back to GetSignedArtifactURL
+		DatabaseID           int64     `json:"databaseId,string"`
+		Name                 string    `json:"name"`
+		Size                 int64     `json:"size,string"`
+		CreatedAt            time.Time `json:"createdAt"` // whole seconds, the toolkit rejects the trimmed fractions json writes
+		Digest               string    `json:"digest"`    // empty until the upload is finalized
+	}
+
+	getSignedArtifactURLRequest struct {
+		WorkflowRunBackendID string
+		Name                 string
+	}
+	getSignedArtifactURLResponse struct {
+		SignedURL string `json:"signedUrl"`
+	}
+
+	createArtifactRequest  = getSignedArtifactURLRequest
+	deleteArtifactRequest  = getSignedArtifactURLRequest
+	deleteArtifactResponse = finalizeArtifactResponse
+)

act/container/docker_proxy.go

@@ -40,7 +40,9 @@ const (
 	jobLabel      = "com.gitea.runner.job."
 	maxCreateBody = 8 << 20
 
-	dockerProxyProbeTimeout = 5 * time.Second
+	dockerProxyProbeTimeout       = 5 * time.Second
+	dockerProxyProbeWorkerTimeout = time.Minute
+	dockerProxyProbeTarget        = "/gitea-runner-probe"
 )
 
 var (
@@ -48,7 +50,7 @@ var (
 	rawStreamPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers/[^/]+/attach|exec/[^/]+/start)$`)
 )
 
-var dockerProxyDirectories, dockerProxyProbes sync.Map
+var dockerProxyProbes sync.Map
 
 type dockerProxyDirectory struct {
 	dir, daemonDir string
@@ -68,7 +70,7 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy {
 	}
 	probeCtx, cancel := context.WithTimeout(ctx, dockerProxyProbeTimeout)
 	defer cancel()
-	cli, err := GetDockerClient(probeCtx)
+	cli, err := client.New(client.FromEnv) // negotiates on first request, so a cached probe costs no round trip
 	if err != nil {
 		return nil
 	}
@@ -83,65 +85,69 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy {
 	}
 	workdir, _ := os.Getwd()
 	key := [3]string{cli.DaemonHost(), workdir, os.TempDir()}
-	cached, _ := dockerProxyDirectories.Load(key)
+	probe, err := probeDockerProxyDirectory(probeCtx, cli, key, socket)
-	directory, seen := cached.(dockerProxyDirectory)
-	seen = seen && os.SameFile(directory.socket, socket) // daemon replacement invalidates cached paths
-	if !seen {
-		directory, seen, err = probeDockerProxyDirectory(probeCtx, cli, key, socket)
-	}
 	if err != nil {
 		common.Logger(ctx).Infof("docker proxy probe failed, jobs get the daemon socket directly: %v", err)
 		return nil
 	}
-	if !seen {
+	if !probe.seen {
 		common.Logger(ctx).Infof("the docker daemon cannot reach the runner's temporary or working directory, jobs get the daemon socket directly")
 		return nil
 	}
 	if ctx.Err() != nil {
 		return nil
 	}
-	proxy, err := StartDockerProxy(daemonSocket, directory.dir, job)
+	proxy, err := StartDockerProxy(daemonSocket, probe.directory.dir, job)
 	if err != nil {
-		dockerProxyDirectories.Delete(key)
+		dockerProxyProbes.CompareAndDelete(key, probe)
 		common.Logger(ctx).Warnf("docker proxy not started, the job gets the daemon socket directly: %v", err)
 		return nil
 	}
-	dockerProxyDirectories.Store(key, directory)
+	proxy.Socket = probe.directory.daemonDir + strings.TrimPrefix(proxy.Socket, probe.directory.dir)
-	proxy.Socket = directory.daemonDir + strings.TrimPrefix(proxy.Socket, directory.dir)
 	return proxy
 }
 
-func probeDockerProxyDirectory(ctx context.Context, cli client.APIClient, key [3]string, socket os.FileInfo) (dockerProxyDirectory, bool, error) {
+func probeDockerProxyDirectory(ctx context.Context, cli client.APIClient, key [3]string, socket os.FileInfo) (*dockerProxyProbe, error) {
 	for {
 		probe := &dockerProxyProbe{directory: dockerProxyDirectory{socket: socket}, done: make(chan struct{})}
 		actual, loaded := dockerProxyProbes.LoadOrStore(key, probe)
 		if loaded {
 			probe, _ = actual.(*dockerProxyProbe)
-			if !os.SameFile(probe.directory.socket, socket) {
+			if !os.SameFile(probe.directory.socket, socket) { // daemon replacement invalidates cached paths
 				dockerProxyProbes.CompareAndDelete(key, probe)
 				continue
 			}
-			select {
-			case <-ctx.Done():
-				return dockerProxyDirectory{}, false, ctx.Err()
-			case <-probe.done:
-			}
-			if ctx.Err() == nil && (errors.Is(probe.err, context.Canceled) || errors.Is(probe.err, context.DeadlineExceeded)) {
-				continue
-			}
 		} else {
-			probe.directory.dir, probe.directory.daemonDir = runnerContainerWorkdir(ctx, cli)
+			go func() { // outlives callers' deadlines so a slow daemon still gets its result cached
-			probe.seen = probe.directory.daemonDir != ""
+				defer cli.Close()
-			if !probe.seen {
+				ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeWorkerTimeout)
-				if probe.directory.dir, probe.err = filepath.Abs(key[2]); probe.err == nil {
+				defer cancel()
-					probe.directory.daemonDir = probe.directory.dir
+				probe.directory.dir, probe.directory.daemonDir = runnerContainerWorkdir(ctx, cli)
-					probe.seen, probe.err = daemonSeesDir(ctx, cli, probe.directory.dir, probe.directory.daemonDir)
+				probe.seen = probe.directory.daemonDir != ""
+				if !probe.seen {
+					if probe.directory.dir, probe.err = filepath.Abs(key[2]); probe.err == nil {
+						probe.directory.daemonDir = probe.directory.dir
+						probe.seen, probe.err = daemonSeesDir(ctx, cli, probe.directory.dir, probe.directory.daemonDir)
+					}
 				}
-			}
+				if probe.seen {
-			dockerProxyProbes.CompareAndDelete(key, probe)
+					dockerProxyProbes.LoadOrStore(key, probe)
-			close(probe.done)
+				} else {
+					dockerProxyProbes.CompareAndDelete(key, probe)
+				}
+				close(probe.done)
+				removeStaleProbeContainers(cli)
+			}()
 		}
-		return probe.directory, probe.seen, probe.err
+		select {
+		case <-ctx.Done():
+			return &dockerProxyProbe{}, ctx.Err()
+		case <-probe.done:
+		}
+		if ctx.Err() == nil && errors.Is(probe.err, context.DeadlineExceeded) {
+			continue
+		}
+		return probe, probe.err
 	}
 }
 
@@ -200,7 +206,7 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str
 		created, err = cli.ContainerCreate(ctx, client.ContainerCreateOptions{
 			Config: &container.Config{Image: image.ID, Cmd: []string{"true"}},
 			HostConfig: &container.HostConfig{Mounts: []mount.Mount{
-				{Type: mount.TypeBind, Source: daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it
+				{Type: mount.TypeBind, Source: daemonDir, Target: dockerProxyProbeTarget, ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it
 			}},
 		})
 		if !cerrdefs.IsNotFound(err) {
@@ -213,7 +219,7 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str
 	if err != nil {
 		return false, err
 	}
-	seen, err := containerSeesMarker(ctx, cli, dir, created.ID, "/gitea-runner-probe")
+	seen, err := containerSeesMarker(ctx, cli, dir, created.ID, dockerProxyProbeTarget)
 	cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), dockerProxyProbeTimeout)
 	defer cancel()
 	if _, removeErr := cli.ContainerRemove(cleanupCtx, created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
@@ -222,6 +228,36 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str
 	return seen, err
 }
 
+// removeStaleProbeContainers removes probe containers left by abandoned creates once any probe in flight when listed has ended.
+func removeStaleProbeContainers(cli client.APIClient) {
+	ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeTimeout)
+	defer cancel()
+	listed, err := cli.ContainerList(ctx, client.ContainerListOptions{
+		All:     true,
+		Filters: make(client.Filters).Add("status", string(container.StateCreated)).Add("volume", dockerProxyProbeTarget),
+	})
+	if err != nil {
+		common.Logger(ctx).Warnf("listing stale docker proxy probe containers failed: %v", err)
+		return
+	}
+	listed.Items = slices.DeleteFunc(listed.Items, func(stale container.Summary) bool {
+		return stale.State != container.StateCreated || !slices.ContainsFunc(stale.Mounts, func(mountPoint container.MountPoint) bool {
+			return mountPoint.Type == mount.TypeBind && !mountPoint.RW && mountPoint.Destination == dockerProxyProbeTarget
+		})
+	})
+	if len(listed.Items) == 0 {
+		return
+	}
+	time.Sleep(dockerProxyProbeWorkerTimeout + dockerProxyProbeTimeout)
+	for _, stale := range listed.Items {
+		ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeTimeout)
+		if err := (&containerReference{cli: cli, id: stale.ID}).remove()(ctx); err != nil {
+			common.Logger(ctx).Warnf("removing stale docker proxy probe container failed: %v", err)
+		}
+		cancel()
+	}
+}
+
 // StartDockerProxy serves a job's docker socket in dir, labelling what the job creates through it.
 func StartDockerProxy(daemonSocket, dir, job string) (*DockerProxy, error) {
 	info, err := os.Stat(daemonSocket)

act/container/docker_proxy_test.go

@@ -94,6 +94,8 @@ func TestDockerProxy(t *testing.T) {
 				return
 			}
 			_, _ = w.Write([]byte(`[{"Id":"probe-image"}]`))
+		case strings.HasSuffix(r.URL.Path, "/containers/json"):
+			_, _ = w.Write([]byte(`[]`))
 		case strings.HasSuffix(r.URL.Path, "/json"):
 			http.NotFound(w, r)
 		case strings.HasSuffix(r.URL.Path, "/archive"):
@@ -235,7 +237,8 @@ func TestDockerProxy(t *testing.T) {
 			t.Skip("automatic proxy discovery requires Linux")
 		}
 		t.Setenv("DOCKER_HOST", "unix://"+daemonSocket)
-		t.Setenv("TMPDIR", shortTempDir(t))
+		tempDir := shortTempDir(t)
+		t.Setenv("TMPDIR", tempDir)
 		require.Nil(t, NewDockerProxy(t.Context(), "failed"))
 		for _, job := range []string{"first", "second"} {
 			proxy := NewDockerProxy(t.Context(), job)
@@ -243,12 +246,10 @@ func TestDockerProxy(t *testing.T) {
 			require.NoError(t, proxy.Close(context.Background()))
 		}
 		assert.EqualValues(t, 2, probes.Load())
-		dockerProxyDirectories.Range(func(key, value any) bool {
+		workdir, err := os.Getwd()
-			directory, _ := value.(dockerProxyDirectory)
+		require.NoError(t, err)
-			directory.socket, _ = os.Stat(t.TempDir())
+		cached, _ := dockerProxyProbes.Load([3]string{"unix://" + daemonSocket, workdir, tempDir})
-			dockerProxyDirectories.Store(key, directory)
+		cached.(*dockerProxyProbe).directory.socket, _ = os.Stat(t.TempDir())
-			return true
-		})
 		proxy := NewDockerProxy(t.Context(), "replaced daemon")
 		require.NotNil(t, proxy)
 		require.NoError(t, proxy.Close(context.Background()))
@@ -476,10 +477,23 @@ func BenchmarkDockerProxy(b *testing.B) {
 
 type probeClient struct {
 	mobyclient.APIClient
-	create func(mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error)
+	create     func(context.Context, mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error)
-	remove func(context.Context, string, mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error)
+	remove     func(context.Context, string, mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error)
-	stat   func(string) error
+	stat       func(string) error
-	mounts []container.MountPoint
+	mounts     []container.MountPoint
+	containers []container.Summary
+}
+
+func (c *probeClient) Close() error {
+	return nil
+}
+
+func (c *probeClient) ContainerList(context.Context, mobyclient.ContainerListOptions) (mobyclient.ContainerListResult, error) {
+	return mobyclient.ContainerListResult{Items: c.containers}, nil
+}
+
+func (c *probeClient) ContainerKill(context.Context, string, mobyclient.ContainerKillOptions) (mobyclient.ContainerKillResult, error) {
+	return mobyclient.ContainerKillResult{}, cerrdefs.ErrConflict
 }
 
 func (c *probeClient) ContainerInspect(context.Context, string, mobyclient.ContainerInspectOptions) (mobyclient.ContainerInspectResult, error) {
@@ -490,8 +504,8 @@ func (c *probeClient) ImageList(context.Context, mobyclient.ImageListOptions) (m
 	return mobyclient.ImageListResult{Items: []image.Summary{{ID: "removed-image"}, {ID: "probe-image"}}}, nil
 }
 
-func (c *probeClient) ContainerCreate(_ context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+func (c *probeClient) ContainerCreate(ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
-	return c.create(opts)
+	return c.create(ctx, opts)
 }
 
 func (c *probeClient) ContainerRemove(ctx context.Context, id string, opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) {
@@ -518,7 +532,7 @@ func TestDockerProxyDirectoryProbes(t *testing.T) {
 				var calls, removals atomic.Int32
 				visible := false
 				cli := &probeClient{
-					create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+					create: func(_ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
 						calls.Add(1)
 						<-release
 						return mobyclient.ContainerCreateResult{ID: "probe"}, createErr
@@ -538,23 +552,18 @@ func TestDockerProxyDirectoryProbes(t *testing.T) {
 				results := make(chan result, 8)
 				for range cap(results) {
 					go func() {
-						_, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
+						probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
-						results <- result{seen, err}
+						results <- result{probe.seen, err}
 					}()
 				}
 				synctest.Wait()
 				assert.EqualValues(t, 1, calls.Load())
-				ctx, cancel := context.WithCancel(t.Context())
-				cancel()
-				_, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket)
-				require.ErrorIs(t, err, context.Canceled)
-				assert.False(t, seen)
 				replacement, err := os.Stat(t.TempDir())
 				require.NoError(t, err)
 				replaced := make(chan result, 1)
 				go func() {
-					_, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, replacement)
+					probe, err := probeDockerProxyDirectory(t.Context(), cli, key, replacement)
-					replaced <- result{seen, err}
+					replaced <- result{probe.seen, err}
 				}()
 				synctest.Wait()
 				assert.EqualValues(t, 2, calls.Load())
@@ -574,11 +583,11 @@ func TestDockerProxyDirectoryProbes(t *testing.T) {
 					assert.Zero(t, removals.Load())
 				}
 				createErr, visible = nil, true
-				directory, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
+				probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
 				require.NoError(t, err)
-				assert.True(t, seen)
+				assert.True(t, probe.seen)
-				assert.Equal(t, dir, directory.dir)
+				assert.Equal(t, dir, probe.directory.dir)
-				assert.Equal(t, dir, directory.daemonDir)
+				assert.Equal(t, dir, probe.directory.daemonDir)
 				assert.EqualValues(t, 3, calls.Load())
 				entries, err := os.ReadDir(dir)
 				require.NoError(t, err)
@@ -586,7 +595,7 @@ func TestDockerProxyDirectoryProbes(t *testing.T) {
 			})
 		})
 	}
-	t.Run("cancelled probe does not cancel another caller", func(t *testing.T) {
+	t.Run("cancelled caller leaves the probe to finish for the others", func(t *testing.T) {
 		synctest.Test(t, func(t *testing.T) {
 			dir := t.TempDir()
 			socket, err := os.Stat(t.TempDir())
@@ -595,49 +604,105 @@ func TestDockerProxyDirectoryProbes(t *testing.T) {
 			ctx, cancel := context.WithCancel(t.Context())
 			defer cancel()
 			release := make(chan struct{})
-			calls := 0
+			var calls, removals atomic.Int32
 			cli := &probeClient{
-				create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+				create: func(ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
-					calls++
+					calls.Add(1)
-					if calls == 1 {
+					<-release
-						<-release
+					return mobyclient.ContainerCreateResult{ID: "probe"}, ctx.Err()
+				},
+				stat: func(_path string) error { return nil },
+				remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) {
+					removals.Add(1)
+					return mobyclient.ContainerRemoveResult{}, nil
+				},
+			}
+			leader, follower, late := make(chan result, 1), make(chan result, 1), make(chan result, 1)
+			go func() {
+				probe, err := probeDockerProxyDirectory(ctx, cli, key, socket)
+				leader <- result{probe.seen, err}
+			}()
+			synctest.Wait()
+			go func() {
+				probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
+				follower <- result{probe.seen, err}
+			}()
+			synctest.Wait()
+			cancel()
+			leaderResult := <-leader
+			require.ErrorIs(t, leaderResult.err, context.Canceled)
+			assert.False(t, leaderResult.seen)
+			go func() {
+				probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
+				late <- result{probe.seen, err}
+			}()
+			synctest.Wait()
+			close(release)
+			for _, results := range []chan result{follower, late} {
+				result := <-results
+				require.NoError(t, result.err)
+				assert.True(t, result.seen)
+			}
+			assert.EqualValues(t, 1, calls.Load())
+			assert.EqualValues(t, 1, removals.Load())
+			_, cached := dockerProxyProbes.Load(key)
+			assert.True(t, cached)
+		})
+	})
+	t.Run("caller joining a probe that ran out of time retries", func(t *testing.T) {
+		synctest.Test(t, func(t *testing.T) {
+			dir := t.TempDir()
+			socket, err := os.Stat(t.TempDir())
+			require.NoError(t, err)
+			var calls atomic.Int32
+			cli := &probeClient{
+				create: func(ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+					if calls.Add(1) == 1 {
+						<-ctx.Done()
 						return mobyclient.ContainerCreateResult{}, ctx.Err()
-					} else {
-						return mobyclient.ContainerCreateResult{ID: "probe"}, nil
 					}
+					return mobyclient.ContainerCreateResult{ID: "probe"}, nil
 				},
 				stat: func(_path string) error { return nil },
 				remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) {
 					return mobyclient.ContainerRemoveResult{}, nil
 				},
 			}
-			leader, follower := make(chan result, 1), make(chan result, 1)
+			probe, err := probeDockerProxyDirectory(t.Context(), cli, [3]string{"daemon", dir, dir}, socket)
-			go func() {
+			require.NoError(t, err)
-				_, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket)
+			assert.True(t, probe.seen)
-				leader <- result{seen, err}
+			assert.EqualValues(t, 2, calls.Load())
-			}()
-			synctest.Wait()
-			go func() {
-				_, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket)
-				follower <- result{seen, err}
-			}()
-			synctest.Wait()
-			assert.Equal(t, 1, calls)
-			cancel()
-			close(release)
-			leaderResult, followerResult := <-leader, <-follower
-			require.ErrorIs(t, leaderResult.err, context.Canceled)
-			assert.False(t, leaderResult.seen)
-			require.NoError(t, followerResult.err)
-			assert.True(t, followerResult.seen)
-			assert.Equal(t, 2, calls)
 		})
 	})
 }
 
+func TestRemoveStaleProbeContainers(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		probeMount := container.MountPoint{Type: mount.TypeBind, Destination: "/gitea-runner-probe"}
+		writableMount := container.MountPoint{Type: mount.TypeBind, Destination: "/gitea-runner-probe", RW: true}
+		var removed []string
+		cli := &probeClient{
+			containers: []container.Summary{
+				{ID: "abandoned", State: container.StateCreated, Mounts: []container.MountPoint{probeMount}},
+				{ID: "foreign", State: container.StateCreated, Mounts: []container.MountPoint{writableMount}},
+				{ID: "started", State: container.StateExited, Mounts: []container.MountPoint{probeMount}},
+			},
+			remove: func(_ctx context.Context, id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) {
+				removed = append(removed, id)
+				return mobyclient.ContainerRemoveResult{}, nil
+			},
+		}
+		start := time.Now()
+		removeStaleProbeContainers(cli)
+		assert.Equal(t, []string{"abandoned"}, removed)
+		assert.GreaterOrEqual(t, time.Since(start), dockerProxyProbeWorkerTimeout+dockerProxyProbeTimeout)
+	})
+}
+
 func TestDaemonSeesDir(t *testing.T) {
 	dir := t.TempDir()
 	markers := make(map[string]bool)
+
 	for _, testCase := range []struct {
 		name      string
 		daemonDir string
@@ -656,7 +721,7 @@ func TestDaemonSeesDir(t *testing.T) {
 			removed := false
 			var attemptedImages []string
 			cli := &probeClient{
-				create: func(opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+				create: func(_ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
 					assert.Equal(t, []mount.Mount{{Type: mount.TypeBind, Source: testCase.daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}}, opts.HostConfig.Mounts)
 					attemptedImages = append(attemptedImages, opts.Config.Image)
 					if opts.Config.Image == "removed-image" {
@@ -692,6 +757,7 @@ func TestDaemonSeesDir(t *testing.T) {
 			assert.Equal(t, []string{"removed-image", "probe-image"}, attemptedImages)
 			assert.Equal(t, testCase.seen, seen)
 			assert.Equal(t, testCase.removed, removed)
+
 			entries, err := os.ReadDir(dir)
 			require.NoError(t, err)
 			assert.Empty(t, entries)

act/container/docker_run_test.go

@@ -980,7 +980,7 @@ func TestDockerCreateKeepsInputEntrypoint(t *testing.T) {
 					RunnerOptions:   "--entrypoint=runner-entrypoint",
 					WorkflowOptions: "--entrypoint=" + optionEntrypoint + " --init",
 				},
-				cli: &probeClient{create: func(opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
+				cli: &probeClient{create: func(_ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) {
 					if len(entrypoint) > 0 {
 						assert.Equal(t, entrypoint, opts.Config.Entrypoint)
 					} else {

act/container/kubernetes.go

@@ -7,6 +7,7 @@ import (
 	"bufio"
 	"bytes"
 	"cmp"
+	"compress/gzip"
 	"context"
 	"crypto/rand"
 	"errors"
@@ -55,6 +56,7 @@ type kubernetesContainer struct {
 	name  string
 	input *NewContainerInput
 	probe map[string]any
+	gzip  bool // the container can gunzip, so copies into it go compressed
 }
 
 type containerStatus struct {
@@ -391,8 +393,11 @@ func (c *kubernetesContainer) Start(_ bool) common.Executor {
 		if err != nil {
 			return err
 		}
-		machine, path, _ := strings.Cut(output, "\n")
+		probed := strings.SplitN(output, "\n", 3)
-		c.pod.arch, c.pod.path = goArchToActionArch(machine), path
+		for len(probed) < 3 {
+			probed = append(probed, "")
+		}
+		c.pod.arch, c.gzip, c.pod.path = goArchToActionArch(probed[0]), probed[1] == "gzip", probed[2]
 		return nil
 	}
 }
@@ -403,7 +408,7 @@ func (p *KubernetesPod) probe(ctx context.Context, container string) (string, er
 	deadline := time.Now().Add(time.Minute)
 	for delay := 100 * time.Millisecond; ; delay = min(2*delay, 2*time.Second) {
 		var stdout bytes.Buffer
-		err := p.client.exec(ctx, p.name, container, []string{"sh", "-c", `printf '%s\n%s' "$(uname -m)" "$PATH"`}, nil, &stdout, io.Discard)
+		err := p.client.exec(ctx, p.name, container, []string{"sh", "-c", `printf '%s\n%s\n%s' "$(uname -m)" "$(command -v gzip >/dev/null && echo gzip)" "$PATH"`}, nil, &stdout, io.Discard)
 		var exitErr ExitCodeError
 		var apiErr *apiError
 		if err == nil || ctx.Err() != nil || errors.As(err, &exitErr) || time.Now().After(deadline) ||
@@ -452,12 +457,29 @@ func (c *kubernetesContainer) Exec(command []string, env map[string]string, _, w
 	}
 }
 
-func (c *kubernetesContainer) extract(ctx context.Context, destPath string, write func(io.Writer) error) error {
+// compressCopyMin is the size from which Copy compresses: below it gzip saves next to nothing on the wire and costs a process in the pod.
+const compressCopyMin = 64 << 10
+
+// extract streams the tar write produces into destPath over exec, which goes through the API server.
+// With compress it is gzipped on the way, as copied actions shrink about fivefold and API server traffic often crosses zones.
+func (c *kubernetesContainer) extract(ctx context.Context, destPath string, compress bool, write func(io.Writer) error) error {
+	compress = compress && c.gzip
 	reader, writer := io.Pipe()
 	defer reader.Close()
-	go func() { writer.CloseWithError(write(writer)) }()
+	go func() {
+		if !compress {
+			writer.CloseWithError(write(writer))
+			return
+		}
+		compressed, _ := gzip.NewWriterLevel(writer, gzip.BestSpeed) // fast enough to not slow copies down, nearly as small as the default level
+		writer.CloseWithError(errors.Join(write(compressed), compressed.Close()))
+	}()
+	script := `mkdir -p "$1" && tar -xf - -C "$1"`
+	if compress {
+		script = `mkdir -p "$1" && gzip -dc | tar -xf - -C "$1"` // gzip -dc rather than tar -z, which GNU tar runs gzip for anyway
+	}
 	var stderr bytes.Buffer
-	if err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", `mkdir -p "$1" && tar -xf - -C "$1"`, "sh", destPath}, reader, io.Discard, &stderr); err != nil {
+	if err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", script, "sh", destPath}, reader, io.Discard, &stderr); err != nil {
 		return fmt.Errorf("extract to %s: %w: %s", destPath, err, strings.TrimSpace(stderr.String()))
 	}
 	return nil
@@ -465,13 +487,17 @@ func (c *kubernetesContainer) extract(ctx context.Context, destPath string, writ
 
 func (c *kubernetesContainer) Copy(destPath string, files ...*FileEntry) common.Executor {
 	return func(ctx context.Context) error {
-		return c.extract(ctx, destPath, func(writer io.Writer) error { return writeFilesTar(ctx, writer, 0, 0, files...) })
+		size := 0
+		for _, file := range files {
+			size += len(file.Body)
+		}
+		return c.extract(ctx, destPath, size >= compressCopyMin, func(writer io.Writer) error { return writeFilesTar(ctx, writer, 0, 0, files...) })
 	}
 }
 
 func (c *kubernetesContainer) CopyDir(destPath, srcPath string, useGitIgnore, skipGitDir bool) common.Executor {
 	return func(ctx context.Context) error {
-		return c.extract(ctx, destPath, func(writer io.Writer) error {
+		return c.extract(ctx, destPath, true, func(writer io.Writer) error { // directories are actions and workspaces, megabytes of compressible sources
 			return writeDirTar(ctx, writer, "/", srcPath, useGitIgnore, skipGitDir, 0, 0)
 		})
 	}

act/container/kubernetes_test.go

@@ -260,6 +260,7 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) {
 	)
 	require.NoError(t, job.Start(false)(ctx))
 	assert.Len(t, cluster.requests["exec"], 3)
+	assert.True(t, pod.job.gzip)
 
 	stepEnv := map[string]string{"INPUT_WHO-TO-GREET": "it's\nme", "SECRET": "hunter2", "UID": "5"}
 	require.NoError(t, job.Exec([]string{"printenv", "SECRET"}, stepEnv, "", "")(ctx))
@@ -276,7 +277,9 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) {
 		assert.NotContains(t, query, "hunter2")
 	}
 
+	lastExec := func() string { return cluster.requests["exec"][len(cluster.requests["exec"])-1] }
 	require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "workflow/event.json", Mode: 0o644, Body: `{"ref":"main"}`})(ctx))
+	assert.NotContains(t, lastExec(), "gzip")
 	archive, err := job.GetContainerArchive(ctx, workdir+"/act/workflow/event.json")
 	require.NoError(t, err)
 	reader := tar.NewReader(archive)
@@ -289,6 +292,12 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) {
 	require.NoError(t, archive.Close())
 	_, err = job.GetContainerArchive(ctx, workdir+"/missing")
 	require.Error(t, err)
+	large := strings.Repeat("a", compressCopyMin)
+	require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "workflow/large", Mode: 0o644, Body: large})(ctx))
+	assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |"))
+	content, err = os.ReadFile(workdir + "/act/workflow/large")
+	require.NoError(t, err)
+	assert.Equal(t, large, string(content))
 
 	actionDir := filepath.Join(t.TempDir(), "action")
 	require.NoError(t, os.MkdirAll(actionDir, 0o755))
@@ -297,7 +306,8 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) {
 	script, err := os.ReadFile(workdir + "/act/actions/sha/index.js")
 	require.NoError(t, err)
 	assert.Equal(t, "main()", string(script))
-	assert.Contains(t, cluster.requests["exec"][len(cluster.requests["exec"])-1], "command="+url.QueryEscape(workdir+"/act/actions/sha/"))
+	assert.Contains(t, lastExec(), "command="+url.QueryEscape(workdir+"/act/actions/sha/"))
+	assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |"))
 
 	pathEnv := map[string]string{}
 	require.NoError(t, job.UpdateFromImageEnv(&pathEnv)(ctx))
@@ -321,6 +331,38 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) {
 	assert.Empty(t, cluster.requests["DELETE /api/v1/namespaces/ci/secrets/new"])
 }
 
+func TestKubernetesCopiesUncompressedWithoutGzip(t *testing.T) {
+	t.Parallel()
+	if runtime.GOOS == "windows" {
+		t.Skip("uses POSIX shell")
+	}
+	ctx := t.Context()
+	workdir, bin := t.TempDir(), t.TempDir()
+	for _, tool := range []string{"uname", "tar", "mkdir"} { // all the job container needs but gzip, as on a minimal image
+		found, err := exec.LookPath(tool)
+		require.NoError(t, err)
+		require.NoError(t, os.Symlink(found, filepath.Join(bin, tool)))
+	}
+	kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"phase":"Running","containerStatuses":[
+		{"name":"job","ready":true,"state":{"running":{}}}]}}`, "PATH="+bin)
+	pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig})
+	require.NoError(t, err)
+	job := pod.JobContainer(&NewContainerInput{Image: "node", WorkingDir: workdir})
+	require.NoError(t, job.Create(nil, nil)(ctx))
+	require.NoError(t, job.Start(false)(ctx))
+	assert.False(t, pod.job.gzip)
+	assert.Equal(t, bin, pod.path)
+
+	actionDir := filepath.Join(t.TempDir(), "action")
+	require.NoError(t, os.MkdirAll(actionDir, 0o755))
+	require.NoError(t, os.WriteFile(filepath.Join(actionDir, "index.js"), []byte("main()"), 0o644))
+	require.NoError(t, job.CopyDir(workdir+"/act/actions/sha/", actionDir+"/", false, false)(ctx))
+	script, err := os.ReadFile(workdir + "/act/actions/sha/index.js")
+	require.NoError(t, err)
+	assert.Equal(t, "main()", string(script))
+	assert.NotContains(t, cluster.requests["exec"][len(cluster.requests["exec"])-1], "gzip")
+}
+
 func TestKubernetesPodFailsBeforeSteps(t *testing.T) {
 	kubeconfig, _ := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"containerStatuses":[
 		{"name":"job","state":{"waiting":{"reason":"ErrImagePull","message":"no such image"}}}]}}`)

act/runner/action.go

@@ -180,9 +180,9 @@ func runActionImpl(step actionStep, actionDir string, remoteAction *remoteAction
 
 		action := step.getActionModel()
 
-		rc.withGithubEnv(ctx, step.getGithubContext(ctx), *step.getEnv())
+		rc.withGithubEnv(ctx, getStepGithubContext(ctx, step), *step.getEnv())
 		populateEnvsFromSavedState(step.getEnv(), step, rc)
-		if err := populateEnvsFromInput(ctx, step.getEnv(), action, rc); err != nil {
+		if err := populateEnvsFromInput(ctx, step, action); err != nil {
 			return err
 		}
 		warnUnexpectedInputs(ctx, action, stepModel.With)
@@ -283,7 +283,7 @@ func (rc *RunContext) dockerActionImage(ctx context.Context, step actionStep, ac
 	}
 	image, ok := job.dockerActionImages[step.getStepModel()]
 	if !ok {
-		image = dockerActionImageTag(step.getGithubContext(ctx).Repository, actionName, localAction) + "-" + strings.ToLower(rand.Text())
+		image = dockerActionImageTag(getStepGithubContext(ctx, step).Repository, actionName, localAction) + "-" + strings.ToLower(rand.Text())
 		if !common.Dryrun(ctx) { // a dry run builds nothing to remove
 			job.dockerActionImages[step.getStepModel()] = image
 		}
@@ -437,7 +437,7 @@ func evalDockerEnv(ctx context.Context, step step, action *model.Action) (*expre
 	for key := range step.getStepModel().With {
 		providedInputs[strings.ToLower(key)] = true
 	}
-	eval := rc.NewExpressionEvaluator(ctx)
+	eval := rc.NewStepExpressionEvaluator(ctx, step)
 	for k, input := range action.Inputs {
 		if providedInputs[strings.ToLower(k)] {
 			continue
@@ -509,8 +509,9 @@ func inputEnvKey(inputID string) string {
 	return "INPUT_" + inputEnvPattern.ReplaceAllString(strings.ToUpper(inputID), "_")
 }
 
-func populateEnvsFromInput(ctx context.Context, env *map[string]string, action *model.Action, rc *RunContext) error {
+func populateEnvsFromInput(ctx context.Context, step step, action *model.Action) error {
-	eval := rc.NewExpressionEvaluator(ctx)
+	env := step.getEnv()
+	eval := step.getRunContext().NewStepExpressionEvaluator(ctx, step)
 	for inputID, input := range action.Inputs {
 		envKey := inputEnvKey(inputID)
 		if _, ok := (*env)[envKey]; !ok {
@@ -652,7 +653,7 @@ func runPreStep(step actionStep) common.Executor {
 		x := action.Runs.Using
 		if !x.IsComposite() {
 			// defaults in pre steps were missing, however provided inputs are available
-			if err := populateEnvsFromInput(ctx, step.getEnv(), action, rc); err != nil {
+			if err := populateEnvsFromInput(ctx, step, action); err != nil {
 				return err
 			}
 		}

act/runner/action_composite.go

@@ -57,9 +57,6 @@ func evaluateCompositeInputAndEnv(ctx context.Context, parent *RunContext, step
 			env[envKey] = value
 		}
 	}
-	gh := step.getGithubContext(ctx)
-	env["GITHUB_ACTION_REPOSITORY"] = gh.ActionRepository
-	env["GITHUB_ACTION_REF"] = gh.ActionRef
 
 	return env, nil
 }
@@ -110,6 +107,7 @@ func newCompositeRunContext(ctx context.Context, parent *RunContext, step action
 		kubernetes:    parent.kubernetes,
 		jobIndex:      parent.jobIndex,
 		jobTotal:      parent.jobTotal,
+		runtimeToken:  parent.runtimeToken,
 	}
 	compositerc.setCompositeActionEnv(env)
 	compositerc.ExprEval = compositerc.NewExpressionEvaluator(ctx)

act/runner/action_composite_test.go

@@ -19,6 +19,31 @@ import (
 )
 
 func TestCompositeActionParity(t *testing.T) {
+	t.Run("input defaults use the step identity", func(t *testing.T) {
+		for _, uses := range []string{"./action", "own/action@v2"} {
+			t.Run(uses, func(t *testing.T) {
+				rc := &RunContext{
+					Config: &Config{PresetGitHubContext: &model.GithubContext{}},
+					Run:    &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}},
+					Env:    map[string]string{"GITHUB_ACTION_REPOSITORY": "other/composite", "GITHUB_ACTION_REF": "v1"},
+				}
+				var currentStep step = &stepActionLocal{RunContext: rc, Step: &model.Step{Uses: uses}, env: map[string]string{}}
+				want := "|"
+				if uses == "own/action@v2" {
+					currentStep = &stepActionRemote{RunContext: rc, Step: &model.Step{Uses: uses}, env: map[string]string{}, remoteAction: mustNewRemoteAction(t, uses)}
+					want = "own/action|v2"
+				}
+				action := &model.Action{Inputs: map[string]model.Input{"identity": {Default: "${{ github.action_repository }}|${{ github.action_ref }}"}}}
+				require.NoError(t, setupEnv(t.Context(), currentStep))
+				require.NoError(t, populateEnvsFromInput(t.Context(), currentStep, action))
+				assert.Equal(t, want, (*currentStep.getEnv())["INPUT_IDENTITY"])
+				_, err := evalDockerEnv(t.Context(), currentStep, action)
+				require.NoError(t, err)
+				assert.Equal(t, want, (*currentStep.getEnv())["identity"])
+			})
+		}
+	})
+
 	t.Run("inherits contexts without leaking inputs", func(t *testing.T) {
 		hook := logtest.NewWithLevel(logging.LevelWarn)
 		ctx := common.WithLogger(t.Context(), logging.New(hook))
@@ -54,7 +79,7 @@ func TestCompositeActionParity(t *testing.T) {
 		assert.NotContains(t, composite.Env, "INPUT_EXTRA")
 
 		nestedEnv := composite.GetEnv()
-		require.NoError(t, populateEnvsFromInput(ctx, &nestedEnv, &model.Action{Inputs: map[string]model.Input{"shared": {Default: "inner-default"}}}, composite))
+		require.NoError(t, populateEnvsFromInput(ctx, &stepActionLocal{RunContext: composite, env: nestedEnv}, &model.Action{Inputs: map[string]model.Input{"shared": {Default: "inner-default"}}}))
 		assert.Equal(t, "inner-default", nestedEnv["INPUT_SHARED"])
 	})
 

act/runner/expression.go

@@ -90,18 +90,18 @@ var hashfiles string
 
 // NewStepExpressionEvaluator creates a new evaluator with the `inputs` of the enclosing workflow or composite action
 func (rc *RunContext) NewStepExpressionEvaluator(ctx context.Context, step step) *ExpressionEvaluator {
-	return rc.newStepExpressionEvaluator(ctx, step, rc.actionInputs, rc.getJobContext())
+	return rc.newStepExpressionEvaluator(ctx, step, rc.actionInputs, rc.getJobContext(), getStepGithubContext(ctx, step))
 }
 
 // NewActionInputsExpressionEvaluator creates a new evaluator with the step's own with: values as `inputs`
 func (rc *RunContext) NewActionInputsExpressionEvaluator(ctx context.Context, step step) *ExpressionEvaluator {
-	return rc.newStepExpressionEvaluator(ctx, step, inputsFromEnv(*step.getEnv()), rc.getJobContext())
+	return rc.newStepExpressionEvaluator(ctx, step, inputsFromEnv(*step.getEnv()), rc.getJobContext(), getStepGithubContext(ctx, step))
 }
 
-func (rc *RunContext) newStepExpressionEvaluator(ctx context.Context, step step, stepInputs map[string]any, jobContext *model.JobContext) *ExpressionEvaluator {
+func (rc *RunContext) newStepExpressionEvaluator(ctx context.Context, step step, stepInputs map[string]any, jobContext *model.JobContext, github *model.GithubContext) *ExpressionEvaluator {
 	// todo: cleanup EvaluationEnvironment creation
 	ee := &exprparser.EvaluationEnvironment{
-		Github:   step.getGithubContext(ctx),
+		Github:   github,
 		Env:      *step.getEnv(),
 		Job:      jobContext,
 		Steps:    rc.StepResults,
@@ -112,7 +112,7 @@ func (rc *RunContext) newStepExpressionEvaluator(ctx context.Context, step step,
 		Needs:    exprparser.NeedsContext(rc.Run),
 		// todo: should be unavailable
 		// but required to interpolate/evaluate the inputs in actions/composite
-		Inputs:    getEvaluatorInputs(rc, stepInputs, rc.getGithubContext(ctx)),
+		Inputs:    getEvaluatorInputs(rc, stepInputs, github),
 		HashFiles: getHashFilesFunction(ctx, rc),
 	}
 	ee.Runner = rc.getRunnerContext(ctx)
@@ -332,5 +332,10 @@ func getWorkflowSecrets(rc *RunContext) map[string]string {
 	if rc.caller != nil {
 		return rc.workflowCallSecrets
 	}
-	return rc.Config.Secrets
+	if rc.runtimeToken == "" {
+		return rc.Config.Secrets
+	}
+	secrets := map[string]string{"ACTIONS_RUNTIME_TOKEN": rc.runtimeToken} // as the daemon exposes its token
+	maps.Copy(secrets, rc.Config.Secrets)
+	return secrets
 }

act/runner/expression_test.go

@@ -504,7 +504,7 @@ jobs:
 	assert.Equal(t, "workflow", env["INPUT_FALLBACK"])
 }
 
-func TestResolveWorkflowCallToleratesMismatchesAndPassesInheritedSecretsVerbatim(t *testing.T) {
+func TestResolveWorkflowCallToleratesMismatchesAndPassesInheritedSecretsAndRuntimeToken(t *testing.T) {
 	callerWorkflow, err := model.ReadWorkflow(strings.NewReader(`
 jobs:
   call:
@@ -527,7 +527,11 @@ jobs:
 	require.NoError(t, err)
 	notCallable, err := model.ReadWorkflow(strings.NewReader("on: push\njobs:\n  test: {}\n"))
 	require.NoError(t, err)
-	config := &Config{Secrets: map[string]string{"token": "${{ github.token }}"}, PresetGitHubContext: &model.GithubContext{}}
+	config := &Config{
+		Secrets:             map[string]string{"token": "${{ github.token }}"},
+		PresetGitHubContext: &model.GithubContext{},
+		RuntimeTokens:       map[*model.Workflow]string{callerWorkflow: "caller-run"},
+	}
 	resolve := func(callerJobID string, workflow *model.Workflow) *RunContext {
 		parent, err := (&runnerImpl{config: config}).newRunContext(t.Context(), &model.Run{Workflow: callerWorkflow, JobID: callerJobID}, nil)
 		require.NoError(t, err)
@@ -539,9 +543,10 @@ jobs:
 	called := resolve("call", callee)
 	assert.Equal(t, map[string]any{"flag": true}, called.workflowCallInputs)
 	assert.Equal(t, map[string]string{"undeclared": "${{ github.token }}"}, called.workflowCallSecrets)
+	assert.Equal(t, "caller-run", called.GetEnv()["ACTIONS_RUNTIME_TOKEN"])
 	inherited := resolve("inherit", notCallable)
 	assert.Empty(t, inherited.workflowCallInputs)
-	assert.Equal(t, config.Secrets, inherited.workflowCallSecrets)
+	assert.Equal(t, map[string]string{"token": "${{ github.token }}", "ACTIONS_RUNTIME_TOKEN": "caller-run"}, inherited.workflowCallSecrets)
 }
 
 func TestJobNameMasksSecrets(t *testing.T) {

act/runner/job_executor.go

@@ -111,7 +111,7 @@ func printPrepareActions(rc *RunContext, preparers []actionPreparer) common.Exec
 			if !ok || strings.Contains(action.Step.Uses, "${{") || action.action != nil {
 				continue
 			}
-			github := action.getGithubContext(common.WithLogger(ctx, logging.New(logging.DiscardHandler)))
+			github := action.RunContext.getGithubContext(common.WithLogger(ctx, logging.New(logging.DiscardHandler)))
 			remote, err := newRemoteAction(action.Step.Uses, github, action.RunContext.Config.GitHubInstance)
 			if err != nil || action.skipsLocalCheckout(remote, github) {
 				continue

act/runner/run_context.go

@@ -81,6 +81,7 @@ type RunContext struct {
 	jobRunDefaults      model.RunDefaults // defaults.run, resolved once at job setup as GitHub does
 	platformImage       string            // container.image or the runs-on pick, resolved once by isEnabled
 	jobIndex, jobTotal  int               // strategy.job-index and job-total, total 0 when unknown
+	runtimeToken        string            // ACTIONS_RUNTIME_TOKEN of the workflow run, shared with its called workflows and composite actions
 
 	jobEnv             map[string]string          // workflow and job env as evaluated for the container and service specs
 	dockerActionImages map[*model.Step]string     // Dockerfile action images built for this job, removed when it ends
@@ -193,6 +194,9 @@ func (rc *RunContext) GetEnv() map[string]string {
 				rc.Env = mergeMaps(rc.Run.Workflow.Env, job.Environment(), rc.Config.Env)
 			}
 		}
+		if rc.runtimeToken != "" {
+			rc.Env["ACTIONS_RUNTIME_TOKEN"] = rc.runtimeToken // as the daemon's Config.Env carries its token
+		}
 	}
 	if !rc.Config.DisableActEnv {
 		rc.Env["ACT"] = "true"
@@ -1662,7 +1666,7 @@ func (rc *RunContext) withGithubEnv(ctx context.Context, github *model.GithubCon
 
 	if rc.Config.ArtifactServerPath != "" {
 		env["ACTIONS_RUNTIME_URL"] = cmp.Or(os.Getenv("ACTIONS_RUNTIME_URL"), "http://"+net.JoinHostPort(rc.Config.ArtifactServerAddr, rc.Config.ArtifactServerPort)+"/")
-		env["ACTIONS_RUNTIME_TOKEN"] = cmp.Or(os.Getenv("ACTIONS_RUNTIME_TOKEN"), "token")
+		env["ACTIONS_RUNTIME_TOKEN"] = cmp.Or(rc.runtimeToken, os.Getenv("ACTIONS_RUNTIME_TOKEN"), "token")
 	}
 
 	if imageOS := rc.imageOS(ctx); imageOS != "" {

act/runner/run_context_test.go

@@ -1631,4 +1631,8 @@ func TestRunContextWithGithubEnvRunnerValues(t *testing.T) {
 		rc.withGithubEnv(ctx, &model.GithubContext{}, env)
 		assert.Equal(t, ci, env["CI"])
 	}
+
+	rc.runtimeToken = "run-token"
+	newTestCompositeRunContext(t, rc).withGithubEnv(ctx, &model.GithubContext{}, env)
+	assert.Equal(t, "run-token", env["ACTIONS_RUNTIME_TOKEN"])
 }

act/runner/runner.go

@@ -41,6 +41,7 @@ type Config struct {
 	Env                           map[string]string                             // env for containers
 	Secrets                       map[string]string                             // list of secrets
 	ExtraMasks                    []string                                      // values to hide that are not in Secrets, such as the proxy password
+	RuntimeTokens                 map[*model.Workflow]string                    // ACTIONS_RUNTIME_TOKEN per workflow run, as each is its own artifact run
 	Vars                          map[string]string                             // list of vars
 	Token                         string                                        // GitHub token
 	InsecureSecrets               bool                                          // switch hiding output when printing to terminal
@@ -355,6 +356,11 @@ func (runner *runnerImpl) newCombinationRunContext(ctx context.Context, run *mod
 		caller:      runner.caller,
 		jobIndex:    index,
 		jobTotal:    total,
+
+		runtimeToken: runner.config.RuntimeTokens[run.Workflow],
+	}
+	if runner.caller != nil {
+		rc.runtimeToken = runner.caller.runContext.runtimeToken
 	}
 	if err := rc.resolveWorkflowCall(ctx); err != nil {
 		return nil, err

act/runner/step.go

@@ -27,12 +27,24 @@ type step interface {
 	post() common.Executor
 
 	getRunContext() *RunContext
-	getGithubContext(ctx context.Context) *model.GithubContext
 	getStepModel() *model.Step
 	getEnv() *map[string]string
 	getIfExpression(context context.Context, stage stepStage) string
 }
 
+func stepActionIdentity(step step) (repository, ref string) {
+	if action, ok := step.(interface{ actionIdentity() (string, string) }); ok {
+		return action.actionIdentity()
+	}
+	return "", ""
+}
+
+func getStepGithubContext(ctx context.Context, step step) *model.GithubContext {
+	github := step.getRunContext().getGithubContext(ctx)
+	github.ActionRepository, github.ActionRef = stepActionIdentity(step)
+	return github
+}
+
 type stepStage int
 
 const (
@@ -92,6 +104,8 @@ func runStepExecutor(step step, stage stepStage, executor common.Executor) commo
 			runStep, err = isStepEnabled(ctx, ifExpression, step, stage)
 		}
 		if err == nil && runStep {
+			env := *step.getEnv()
+			env["GITHUB_ACTION_REPOSITORY"], env["GITHUB_ACTION_REF"] = stepActionIdentity(step)
 			err = setupInputs(ctx, step)
 		}
 		if err != nil {
@@ -253,7 +267,7 @@ func setupEnv(ctx context.Context, step step) error {
 	env := step.getEnv()
 	// container env is the image's baseline, which job env and $GITHUB_ENV override
 	mergeIntoMap(step, env, rc.containerSpec.Env, rc.GetEnv())
-	rc.withGithubEnv(ctx, step.getGithubContext(ctx), *env)
+	rc.withGithubEnv(ctx, getStepGithubContext(ctx, step), *env)
 
 	var err error
 	exprEval := rc.NewExpressionEvaluator(ctx)
@@ -302,7 +316,7 @@ func isStepEnabled(ctx context.Context, expr string, step step, stage stepStage)
 	if rc.Parent != nil && stage == stepStageMain && jobContext.Status != "cancelled" {
 		jobContext.Status = rc.ownStatus()
 	}
-	runStep, err := EvalBool(ctx, rc.newStepExpressionEvaluator(ctx, step, rc.actionInputs, jobContext), expr, defaultStatusCheck)
+	runStep, err := EvalBool(ctx, rc.newStepExpressionEvaluator(ctx, step, rc.actionInputs, jobContext, rc.getGithubContext(ctx)), expr, defaultStatusCheck)
 	if err != nil {
 		return false, fmt.Errorf("if-expression %q evaluation failed: %s", expr, err)
 	}

act/runner/step_action_builtin.go

@@ -39,7 +39,7 @@ func (sab *stepActionBuiltin) main() common.Executor {
 		rc.ApplyExtraPath(ctx, &sab.env)
 		return sab.run(ctx, &action.Context{
 			Container:       rc.JobContainer,
-			Github:          rc.getGithubContext(ctx),
+			Github:          getStepGithubContext(ctx, sab),
 			Inputs:          sab.Step.With,
 			HostWorkdir:     rc.Config.Workdir,
 			Workspace:       rc.JobContainer.ToContainerPath(rc.Config.Workdir),
@@ -58,10 +58,6 @@ func (sab *stepActionBuiltin) post() common.Executor { return common.NewPipeline
 
 func (sab *stepActionBuiltin) getRunContext() *RunContext { return sab.RunContext }
 
-func (sab *stepActionBuiltin) getGithubContext(ctx context.Context) *model.GithubContext {
-	return sab.RunContext.getGithubContext(ctx)
-}
-
 func (sab *stepActionBuiltin) getStepModel() *model.Step { return sab.Step }
 
 func (sab *stepActionBuiltin) getEnv() *map[string]string { return &sab.env }

act/runner/step_action_local.go

@@ -95,10 +95,6 @@ func (sal *stepActionLocal) getRunContext() *RunContext {
 	return sal.RunContext
 }
 
-func (sal *stepActionLocal) getGithubContext(ctx context.Context) *model.GithubContext {
-	return sal.getRunContext().getGithubContext(ctx)
-}
-
 func (sal *stepActionLocal) getStepModel() *model.Step {
 	return sal.Step
 }

act/runner/step_action_local_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"go.yaml.in/yaml/v4"
 )
 
 type stepActionMocks struct {
@@ -53,7 +54,7 @@ func TestStepActionLocalTest(t *testing.T) {
 		readAction: salm.readAction,
 		runAction:  salm.runAction,
 		RunContext: &RunContext{
-			StepResults: map[string]*model.StepResult{},
+			StepResults: map[string]*model.StepResult{"previous": {Outputs: map[string]string{"value": "previous"}}},
 			ExprEval:    &expressionEvaluator{},
 			Config: &Config{
 				Workdir: "/tmp",
@@ -68,25 +69,42 @@ func TestStepActionLocalTest(t *testing.T) {
 				},
 			},
 			JobContainer: cm,
+			Env:          map[string]string{"GITHUB_ACTION_REPOSITORY": "other/action", "GITHUB_ACTION_REF": "v1"},
+			Matrix:       map[string]any{"value": "matrix"},
+			actionInputs: map[string]any{"shared": "input"},
 		},
 		Step: &model.Step{
 			ID:   "1",
 			Uses: "./path/to/action",
+			If: yaml.Node{Value: "github.action_repository == 'other/action' && github.action_ref == 'v1' && " +
+				"env.STEP_VALUE == 'step' && env.GITHUB_ACTION_REPOSITORY == 'forged/repo' && env.GITHUB_ACTION_REF == 'forged-ref' && " +
+				"steps.previous.outputs.value == 'previous' && inputs.shared == 'input' && matrix.value == 'matrix'"},
 		},
 	}
+	require.NoError(t, sal.Step.Env.Encode(map[string]string{"STEP_VALUE": "step", "GITHUB_ACTION_REPOSITORY": "forged/repo", "GITHUB_ACTION_REF": "forged-ref"}))
+	require.NoError(t, sal.Step.RawWith.Encode(map[string]string{"identity": "${{ github.action_repository }}|${{ github.action_ref }}"}))
 
 	salm.On("readAction", sal.Step, filepath.Clean("/tmp/path/to/action"), "", mock.Anything, mock.Anything).
 		Return(&model.Action{}, nil)
 
 	expectStepFileCommands(t, cm)
 
-	salm.On("runAction", sal, filepath.Clean("/tmp/path/to/action"), (*remoteAction)(nil)).Return(noopExecutor)
+	salm.On("runAction", sal, filepath.Clean("/tmp/path/to/action"), (*remoteAction)(nil)).Return(func(ctx context.Context) error {
+		sal.RunContext.setOutput(ctx, map[string]string{"name": "value"}, "ran")
+		return nil
+	})
 
 	err := sal.pre()(ctx)
 	require.NoError(t, err)
 
 	err = sal.main()(ctx)
 	require.NoError(t, err)
+	assert.Empty(t, sal.env["GITHUB_ACTION_REPOSITORY"])
+	assert.Empty(t, sal.env["GITHUB_ACTION_REF"])
+	assert.Equal(t, "|", sal.Step.With["identity"])
+	assert.Equal(t, "ran", sal.RunContext.StepResults["1"].Outputs["value"])
+	assert.Equal(t, "other/action", sal.RunContext.Env["GITHUB_ACTION_REPOSITORY"])
+	assert.Equal(t, "v1", sal.RunContext.Env["GITHUB_ACTION_REF"])
 
 	cm.AssertExpectations(t)
 	salm.AssertExpectations(t)

act/runner/step_action_remote.go

@@ -75,7 +75,7 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
 		}
 		sar.Step.Uses = uses
 
-		github := sar.getGithubContext(ctx) // read before remoteAction is set, so `$/` resolves against the enclosing action
+		github := sar.RunContext.getGithubContext(ctx)
 		if sar.remoteAction, err = newRemoteAction(sar.Step.Uses, github, sar.RunContext.Config.GitHubInstance); err != nil {
 			return err
 		}
@@ -153,7 +153,7 @@ func (sar *stepActionRemote) main() common.Executor {
 	return common.NewPipelineExecutor(
 		sar.prepareActionExecutor(),
 		runStepExecutor(sar, stepStageMain, func(ctx context.Context) error {
-			github := sar.getGithubContext(ctx)
+			github := getStepGithubContext(ctx, sar)
 			if sar.skipsLocalCheckout(sar.remoteAction, github) {
 				printRunActionHeader(ctx, sar.Step, nil, sar.env, sar.RunContext)
 				if sar.RunContext.Config.BindWorkdir {
@@ -181,17 +181,11 @@ func (sar *stepActionRemote) getRunContext() *RunContext {
 	return sar.RunContext
 }
 
-func (sar *stepActionRemote) getGithubContext(ctx context.Context) *model.GithubContext {
+func (sar *stepActionRemote) actionIdentity() (string, string) {
-	ghc := sar.getRunContext().getGithubContext(ctx)
+	if sar.remoteAction == nil {
-
+		return "", ""
-	// extend github context if we already have an initialized remoteAction
-	remoteAction := sar.remoteAction
-	if remoteAction != nil {
-		ghc.ActionRepository = fmt.Sprintf("%s/%s", remoteAction.Org, remoteAction.Repo)
-		ghc.ActionRef = remoteAction.Ref
 	}
-
+	return sar.remoteAction.Org + "/" + sar.remoteAction.Repo, sar.remoteAction.Ref
-	return ghc
 }
 
 func (sar *stepActionRemote) getStepModel() *model.Step {
@@ -205,7 +199,7 @@ func (sar *stepActionRemote) getEnv() *map[string]string {
 func (sar *stepActionRemote) getIfExpression(ctx context.Context, stage stepStage) string {
 	switch stage {
 	case stepStagePre:
-		github := sar.getGithubContext(ctx)
+		github := getStepGithubContext(ctx, sar)
 		if sar.skipsLocalCheckout(sar.remoteAction, github) {
 			// skip local checkout pre step
 			return "false"

act/runner/step_action_remote_test.go

@@ -76,7 +76,7 @@ func TestStepActionRemote(t *testing.T) {
 		status          model.StepStatus
 		runError        error
 	}{
-		{name: "run-successful", status: model.StepStatusSuccess},
+		{name: "run-successful", condition: "github.action_repository == 'other/composite' && github.action_ref == 'v2'", status: model.StepStatusSuccess},
 		{name: "run-skipped", condition: "false", status: model.StepStatusSkipped},
 		{name: "run-error", status: model.StepStatusFailure, runError: assert.AnError},
 	} {
@@ -110,6 +110,7 @@ func TestStepActionRemote(t *testing.T) {
 					},
 					StepResults:  map[string]*model.StepResult{},
 					JobContainer: cm,
+					Env:          map[string]string{"GITHUB_ACTION_REPOSITORY": "other/composite", "GITHUB_ACTION_REF": "v2"},
 				},
 				Step:       &model.Step{ID: "step", Uses: "remote/action@v1", If: yaml.Node{Value: tt.condition}},
 				readAction: sarm.readAction,
@@ -127,6 +128,8 @@ func TestStepActionRemote(t *testing.T) {
 			require.NoError(t, sar.pre()(ctx))
 			assert.Equal(t, tt.runError, sar.main()(ctx))
 			assert.True(t, clonedAction)
+			assert.Equal(t, "remote/action", sar.env["GITHUB_ACTION_REPOSITORY"])
+			assert.Equal(t, "v1", sar.env["GITHUB_ACTION_REF"])
 			assert.Equal(t, &model.StepResult{Conclusion: tt.status, Outcome: tt.status, Outputs: map[string]string{}}, sar.RunContext.StepResults["step"])
 
 			sarm.AssertExpectations(t)
@@ -182,7 +185,7 @@ func TestStepActionRemotePrepare(t *testing.T) {
 	require.NoError(t, err)
 	root, err := prepare(job, "org/repo@v1")
 	require.NoError(t, err)
-	nested, err := prepare(&RunContext{Config: job.Config, Run: job.Run, JobContainer: job.JobContainer, Parent: job}, "org/repo/save@v1")
+	nested, err := prepare(&RunContext{Config: job.Config, Run: job.Run, JobContainer: job.JobContainer, Parent: job, Env: map[string]string{"GITHUB_ACTION_REPOSITORY": "org/repo", "GITHUB_ACTION_REF": "v1"}}, "$/save")
 	require.NoError(t, err)
 	other, err := prepare(job, "org/repo@v2")
 	require.NoError(t, err)
@@ -346,9 +349,6 @@ func Test_newRemoteActionSelfRepo(t *testing.T) {
 			},
 		},
 		{name: "unknown instance", action: "$/.gitea/actions/build", github: &model.GithubContext{Repository: "owner/workflow-repo", Sha: "abc123"}},
-		{name: "empty path", action: "$/", github: workflow},
-		{name: "ref suffix is not allowed", action: "$/.gitea/actions/build@v1", github: workflow},
-		{name: "path traversal", action: "$/../escape", github: workflow},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

act/runner/step_docker.go

@@ -35,10 +35,6 @@ func (sd *stepDocker) getRunContext() *RunContext {
 	return sd.RunContext
 }
 
-func (sd *stepDocker) getGithubContext(ctx context.Context) *model.GithubContext {
-	return sd.getRunContext().getGithubContext(ctx)
-}
-
 func (sd *stepDocker) getStepModel() *model.Step {
 	return sd.Step
 }

act/runner/step_docker_test.go

@@ -14,6 +14,7 @@ import (
 	"gitea.dev/actionslib/pkg/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.yaml.in/yaml/v4"
 )
 
 func TestStepDockerMain(t *testing.T) {
@@ -57,13 +58,16 @@ func TestStepDockerMain(t *testing.T) {
 				},
 			},
 			JobContainer: cm,
+			Env:          map[string]string{"GITHUB_ACTION_REPOSITORY": "other/composite", "GITHUB_ACTION_REF": "v1"},
 		},
 		Step: &model.Step{
 			ID:               "1",
 			Uses:             "docker://node:14",
 			WorkingDirectory: "workdir",
+			If:               yaml.Node{Value: "github.action_repository == 'other/composite' && github.action_ref == 'v1'"},
 		},
 	}
+	require.NoError(t, sd.Step.RawWith.Encode(map[string]string{"args": "identity:${{ github.action_repository }}:${{ github.action_ref }}"}))
 	sd.RunContext.ExprEval = sd.RunContext.NewExpressionEvaluator(ctx)
 
 	cm.On("Pull", false).Return(noopExecutor)
@@ -80,6 +84,11 @@ func TestStepDockerMain(t *testing.T) {
 	require.NoError(t, sd.main()(ctx))
 
 	assert.Equal(t, "node:14", input.Image)
+	assert.Equal(t, []string{"identity::"}, input.Cmd)
+	assert.Contains(t, input.Env, "GITHUB_ACTION_REPOSITORY=")
+	assert.Contains(t, input.Env, "GITHUB_ACTION_REF=")
+	assert.Equal(t, "other/composite", sd.RunContext.Env["GITHUB_ACTION_REPOSITORY"])
+	assert.Equal(t, "v1", sd.RunContext.Env["GITHUB_ACTION_REF"])
 
 	assert.Empty(t, input.Username)
 	assert.Empty(t, input.Password)

act/runner/step_run.go

@@ -222,10 +222,6 @@ func (sr *stepRun) getRunContext() *RunContext {
 	return sr.RunContext
 }
 
-func (sr *stepRun) getGithubContext(ctx context.Context) *model.GithubContext {
-	return sr.getRunContext().getGithubContext(ctx)
-}
-
 func (sr *stepRun) getStepModel() *model.Step {
 	return sr.Step
 }

act/runner/step_run_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"go.yaml.in/yaml/v4"
 )
 
 type shellContainerMock struct{ *containerMock }
@@ -36,7 +37,6 @@ func TestStepRun(t *testing.T) {
 	sr := &stepRun{
 		RunContext: &RunContext{
 			StepResults: map[string]*model.StepResult{},
-			ExprEval:    &expressionEvaluator{},
 			Config:      &Config{Env: testCheckoutEnv()},
 			Run: &model.Run{
 				JobID: "1",
@@ -48,19 +48,27 @@ func TestStepRun(t *testing.T) {
 			},
 			JobContainer:   cm,
 			jobRunDefaults: model.RunDefaults{Shell: "bash"},
+			Env:            map[string]string{"GITHUB_ACTION_REPOSITORY": "other/composite", "GITHUB_ACTION_REF": "v1"},
 		},
 		Step: &model.Step{
 			ID:               "1",
-			Run:              "cmd",
+			Run:              "cmd${{ github.action_repository }}${{ github.action_ref }}",
 			WorkingDirectory: "workdir",
+			If:               yaml.Node{Value: "github.action_repository == 'other/composite' && github.action_ref == 'v1'"},
 		},
 	}
+	sr.RunContext.ExprEval = sr.RunContext.NewExpressionEvaluator(t.Context())
+	require.NoError(t, sr.Step.Env.Encode(map[string]string{"GITHUB_ACTION_REPOSITORY": "forged/repo", "GITHUB_ACTION_REF": "forged-ref"}))
 
 	cm.On("Copy", "/var/run/act", []*container.FileEntry{fileEntry}).Return(noopExecutor)
 	cm.On("Exec", []string{"bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "/var/run/act/workflow/1.sh"}, mock.AnythingOfType("map[string]string"), "", "workdir").Return(noopExecutor)
 
 	expectStepFileCommands(t, cm)
 	require.NoError(t, sr.main()(t.Context()))
+	assert.Empty(t, sr.env["GITHUB_ACTION_REPOSITORY"])
+	assert.Empty(t, sr.env["GITHUB_ACTION_REF"])
+	assert.Equal(t, "other/composite", sr.RunContext.Env["GITHUB_ACTION_REPOSITORY"])
+	assert.Equal(t, "v1", sr.RunContext.Env["GITHUB_ACTION_REF"])
 
 	cm.AssertExpectations(t)
 }

e2e/cli_test.go

@@ -0,0 +1,312 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"slices"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+type cliRunner struct {
+	api     *GiteaAPI
+	repo    string
+	bin     string
+	cfgFile string
+	dir     string
+}
+
+// newCLIRunner registers the built binary against a fresh repo with a single docker label.
+func newCLIRunner(t *testing.T, label, extraRunnerConfig string) *cliRunner {
+	t.Helper()
+	api, repo := newScenario(t)
+	token, err := fixture.RegistrationToken(t.Context(), repo)
+	if err != nil {
+		t.Fatalf("get registration token: %v", err)
+	}
+
+	dir := t.TempDir()
+	bin, err := runnerBinary()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cfg := fmt.Sprintf(`runner:
+  file: %s
+  fetch_interval: 250ms
+  fetch_interval_max: 250ms
+  state_report_interval: 500ms
+  log_report_interval: 500ms
+%scache:
+  dir: %s
+`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache"))
+	if fixture.network != "" {
+		cfg += "container:\n  network: " + fixture.network + "\n"
+	}
+	c := &cliRunner{api: api, repo: repo, bin: bin, cfgFile: filepath.Join(dir, "config.yaml"), dir: dir}
+	if err := os.WriteFile(c.cfgFile, []byte(cfg), 0o600); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+
+	out, err := exec.CommandContext(t.Context(), c.bin, "register", "--no-interactive", "-c", c.cfgFile,
+		"--instance", fixture.baseURL, "--token", token, "--name", repo,
+		"--labels", label+":docker://"+os.Getenv("E2E_JOB_IMAGE")).CombinedOutput()
+	if err != nil {
+		t.Fatalf("register: %v\n%s", err, out)
+	}
+	if _, err := os.Stat(filepath.Join(dir, ".runner")); err != nil {
+		t.Fatalf("register did not write the runner file: %v\n%s", err, out)
+	}
+	return c
+}
+
+type cliDaemon struct {
+	cmd    *exec.Cmd
+	log    *syncBuffer
+	exited chan error
+}
+
+func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon {
+	t.Helper()
+	d := &cliDaemon{log: &syncBuffer{}, exited: make(chan error, 1)}
+	d.cmd = exec.Command(c.bin, append([]string{"daemon", "-c", c.cfgFile}, args...)...)
+	d.cmd.Stdout, d.cmd.Stderr = d.log, d.log
+	if err := d.cmd.Start(); err != nil {
+		t.Fatalf("start daemon: %v", err)
+	}
+	go func() { d.exited <- d.cmd.Wait() }()
+	t.Cleanup(func() {
+		_ = d.cmd.Process.Kill() // no-op once the daemon has exited
+	})
+	return d
+}
+
+// wait returns the daemon's exit error, failing the test when it is still running after timeout.
+func (d *cliDaemon) wait(t *testing.T, timeout time.Duration) error {
+	t.Helper()
+	select {
+	case err := <-d.exited:
+		return err
+	case <-time.After(timeout):
+		t.Fatalf("daemon still running after %s\n%s", timeout, d.log)
+		return nil
+	}
+}
+
+func (d *cliDaemon) terminate(t *testing.T) {
+	t.Helper()
+	if err := d.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+		t.Fatalf("signal daemon: %v", err)
+	}
+}
+
+// requireSuccess waits for the run and dumps the daemon output next to the job logs on failure.
+func (d *cliDaemon) requireSuccess(t *testing.T, c *cliRunner, runID int64) {
+	t.Helper()
+	completed, err := c.api.WaitForRunConclusion(t.Context(), c.repo, runID, runTimeout)
+	if err != nil || completed.Conclusion != "success" {
+		t.Logf("daemon output:\n%s", d.log)
+		requireSuccess(t, c.api, c.repo, runID)
+	}
+}
+
+// testCLIRegisterDaemon drives the built binary the way a user does: register, then daemon.
+func testCLIRegisterDaemon(t *testing.T) {
+	t.Parallel()
+	c := newCLIRunner(t, "e2e-cli", "")
+	d := c.startDaemon(t)
+
+	pushWorkflow(t, c.api, c.repo, "cli.yml")
+	wfRun := waitForRun(t, c.api, c.repo)
+	d.requireSuccess(t, c, wfRun.ID)
+	if logs := runLogs(t, c.api, c.repo, wfRun.ID); commandRow(logs, "e2e-cli-job-ran") == "" {
+		t.Errorf("job logs are missing the marker:\n%s", logs)
+	}
+
+	d.terminate(t)
+	if err := d.wait(t, 30*time.Second); err != nil {
+		t.Errorf("daemon did not exit cleanly on SIGTERM: %v\n%s", err, d.log)
+	}
+}
+
+// testCLIOnce checks that --once takes a single job and then exits on its own.
+func testCLIOnce(t *testing.T) {
+	t.Parallel()
+	c := newCLIRunner(t, "e2e-once", "")
+	d := c.startDaemon(t, "--once")
+
+	pushWorkflow(t, c.api, c.repo, "cli_once.yml")
+	wfRun := waitForRun(t, c.api, c.repo)
+	if err := d.wait(t, runTimeout); err != nil {
+		t.Errorf("daemon --once exited with %v\n%s", err, d.log)
+	}
+
+	jobs, err := c.api.Jobs(t.Context(), c.repo, wfRun.ID)
+	if err != nil {
+		t.Fatalf("list jobs: %v", err)
+	}
+	var succeeded, pending int
+	for _, job := range jobs {
+		switch {
+		case job.Conclusion == "success":
+			succeeded++
+		case job.Status != "in_progress" && job.Status != "completed":
+			pending++
+		}
+	}
+	if succeeded != 1 || pending != 1 {
+		t.Errorf("got %d succeeded and %d waiting jobs, want 1 and 1: %+v\n%s", succeeded, pending, jobs, d.log)
+	}
+}
+
+// testCLILabelsOverride checks that daemon --labels replaces the labels given at registration.
+func testCLILabelsOverride(t *testing.T) {
+	t.Parallel()
+	c := newCLIRunner(t, "e2e-registered", "")
+	d := c.startDaemon(t, "--labels", "e2e-override:docker://"+os.Getenv("E2E_JOB_IMAGE"))
+
+	pushWorkflow(t, c.api, c.repo, "cli_labels.yml")
+	wfRun := waitForRun(t, c.api, c.repo)
+	d.requireSuccess(t, c, wfRun.ID)
+
+	runners, err := c.api.Runners(t.Context(), c.repo)
+	if err != nil {
+		t.Fatalf("list runners: %v", err)
+	}
+	for _, runner := range runners {
+		if runner.Name != c.repo {
+			continue
+		}
+		var labels []string
+		for _, label := range runner.Labels {
+			labels = append(labels, label.Name)
+		}
+		if !slices.Equal(labels, []string{"e2e-override"}) {
+			t.Errorf("runner has labels %v on the server, want only the --labels override", labels)
+		}
+		return
+	}
+	t.Errorf("runner %q not found among %+v", c.repo, runners)
+}
+
+// testCLIShutdownDrains checks that SIGTERM lets a running job finish within shutdown_timeout.
+func testCLIShutdownDrains(t *testing.T) {
+	t.Parallel()
+	c := newCLIRunner(t, "e2e-shutdown", "  shutdown_timeout: 1m\n")
+	d := c.startDaemon(t)
+
+	pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml")
+	wfRun := waitForRun(t, c.api, c.repo)
+	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started")
+
+	d.terminate(t)
+	if err := d.wait(t, runTimeout); err != nil {
+		t.Errorf("daemon did not exit cleanly: %v\n%s", err, d.log)
+	}
+	if strings.Contains(d.log.String(), "cancelled in progress jobs") {
+		t.Errorf("the running job was cancelled instead of drained\n%s", d.log)
+	}
+	d.requireSuccess(t, c, wfRun.ID)
+	if logs := runLogs(t, c.api, c.repo, wfRun.ID); commandRow(logs, "e2e-shutdown-finished") == "" {
+		t.Errorf("job did not run to the end:\n%s", logs)
+	}
+}
+
+// testCLIShutdownCancels checks that a job still running at shutdown_timeout is cancelled.
+func testCLIShutdownCancels(t *testing.T) {
+	t.Parallel()
+	c := newCLIRunner(t, "e2e-shutdown", "  shutdown_timeout: 1s\n")
+	d := c.startDaemon(t)
+
+	pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml")
+	wfRun := waitForRun(t, c.api, c.repo)
+	waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started")
+
+	d.terminate(t)
+	if err := d.wait(t, runTimeout); err != nil {
+		t.Errorf("daemon did not exit cleanly after cancelling the job: %v\n%s", err, d.log)
+	}
+	if !strings.Contains(d.log.String(), "cancelled in progress jobs") {
+		t.Errorf("daemon did not report cancelling the running job\n%s", d.log)
+	}
+	completed, err := c.api.WaitForRunConclusion(t.Context(), c.repo, wfRun.ID, time.Minute)
+	if err != nil {
+		t.Fatalf("cancelled run was never finalized: %v", err)
+	}
+	if completed.Conclusion == "success" {
+		t.Errorf("run concluded success although the job was cancelled")
+	}
+}
+
+func testExecArtifacts(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	bin, err := runnerBinary()
+	if err != nil {
+		t.Fatal(err)
+	}
+	workflow, err := os.ReadFile("testdata/workflows/exec_artifact.yml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	workflows := t.TempDir()
+	for _, name := range []string{"a.yml", "b.yml"} {
+		if err := os.WriteFile(filepath.Join(workflows, name), workflow, 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	workdir := t.TempDir()
+	for _, args := range [][]string{
+		{"init", "-q"},
+		{"remote", "add", "origin", "https://example.com/e2e/exec.git"},
+		{"-c", "user.name=e2e", "-c", "[email protected]", "commit", "-q", "--allow-empty", "-m", "init"},
+	} {
+		if out, err := exec.CommandContext(ctx, "git", append([]string{"-C", workdir}, args...)...).CombinedOutput(); err != nil {
+			t.Fatalf("git %v: %v\n%s", args, err, out)
+		}
+	}
+
+	artifacts := t.TempDir()
+	args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts}
+	if fixture.network != "" {
+		args = append(args, "--network", fixture.network)
+	}
+	cmd := exec.CommandContext(ctx, bin, args...)
+	cmd.Env = slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") })
+	if out, err := cmd.CombinedOutput(); err != nil {
+		t.Fatalf("exec: %v\n%s", err, out)
+	}
+	if blobs, _ := filepath.Glob(filepath.Join(artifacts, "v4", "*", "*", "blob")); len(blobs) != 2 {
+		t.Fatalf("want one v4 artifact blob per workflow run, got %v", blobs)
+	}
+}

e2e/compatibility_test.go

@@ -30,8 +30,20 @@ func TestCompatibility(t *testing.T) {
 	t.Run("cancellation_and_log_streaming", testRunCancellation)
 	t.Run("dispatch", testWorkflowDispatch)
 	t.Run("builtin_checkout", testBuiltinCheckout)
+	t.Run("exec_artifacts", testExecArtifacts)
 	t.Run("moving_action_ref", testMovingActionRef)
 	t.Run("ephemeral", testEphemeralRunner)
+	t.Run("failure_paths", testFailurePaths)
+	t.Run("cli_register_daemon", testCLIRegisterDaemon)
+	t.Run("cli_once", testCLIOnce)
+	t.Run("cli_labels_override", testCLILabelsOverride)
+	t.Run("cli_shutdown_drains", testCLIShutdownDrains)
+	t.Run("cli_shutdown_cancels", testCLIShutdownCancels)
+	t.Run("host_mode", testHostMode)
+	t.Run("reusable_workflow", testReusableWorkflow)
+	t.Run("label_routing", testLabelRouting)
+	t.Run("capacity", testCapacity)
+	t.Run("reconnect", testReconnect)
 	t.Run("kubernetes", testKubernetes)
 }
 

e2e/failure_test.go

@@ -0,0 +1,59 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import "testing"
+
+func testFailurePaths(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	api, repo := newScenario(t)
+	pushWorkflow(t, api, repo, "failure.yml")
+
+	wfRun := waitForRun(t, api, repo)
+	completed, err := api.WaitForRunConclusion(ctx, repo, wfRun.ID, runTimeout)
+	if err != nil {
+		dumpRunLogs(t, api, repo, wfRun.ID)
+		t.Fatalf("wait for run: %v", err)
+	}
+	if completed.Conclusion != "failure" {
+		t.Errorf("run concluded %q, want failure", completed.Conclusion)
+	}
+
+	jobs, err := api.Jobs(ctx, repo, wfRun.ID)
+	if err != nil {
+		t.Fatalf("list jobs: %v", err)
+	}
+	want := map[string]string{
+		"fail":      "failure",
+		"tolerated": "success",
+		"dependent": "skipped",
+		"timeout":   "failure",
+	}
+	for _, job := range jobs {
+		if w, ok := want[job.Name]; ok && job.Conclusion != w {
+			t.Errorf("job %q concluded %q, want %q", job.Name, job.Conclusion, w)
+		}
+		delete(want, job.Name)
+	}
+	for name := range want {
+		t.Errorf("job %q not found", name)
+	}
+
+	logs := runLogs(t, api, repo, wfRun.ID)
+	for _, marker := range []string{"e2e-before-fail", "e2e-failure-cleanup", "e2e-tolerated-next", "e2e-timeout-started"} {
+		if commandRow(logs, marker) != marker {
+			t.Errorf("logs are missing %q", marker)
+		}
+	}
+	if commandRow(logs, "e2e-skipped-step") != "" {
+		t.Error("a success() step ran after a failed step")
+	}
+	if t.Failed() {
+		t.Logf("full stored logs:\n%s", logs)
+	}
+}

e2e/gitea_api.go

@@ -37,6 +37,14 @@ type ActionJob struct {
 	Name       string `json:"name"`
 	Status     string `json:"status"`
 	Conclusion string `json:"conclusion"`
+	RunnerName string `json:"runner_name"`
+}
+
+type ActionRunner struct {
+	Name   string `json:"name"`
+	Labels []struct {
+		Name string `json:"name"`
+	} `json:"labels"`
 }
 
 func (a *GiteaAPI) CreateRepo(ctx context.Context, name string) error {
@@ -146,6 +154,17 @@ func (a *GiteaAPI) Jobs(ctx context.Context, repo string, runID int64) ([]Action
 	return resp.Jobs, nil
 }
 
+func (a *GiteaAPI) Runners(ctx context.Context, repo string) ([]ActionRunner, error) {
+	var resp struct {
+		Runners []ActionRunner `json:"runners"`
+	}
+	url := fmt.Sprintf("/api/v1/repos/%s/%s/actions/runners", giteaAdminUser, repo)
+	if err := a.doJSON(ctx, http.MethodGet, url, nil, &resp); err != nil {
+		return nil, err
+	}
+	return resp.Runners, nil
+}
+
 func (a *GiteaAPI) JobLogs(ctx context.Context, repo string, jobID int64) (string, error) {
 	url := fmt.Sprintf("/api/v1/repos/%s/%s/actions/jobs/%d/logs", giteaAdminUser, repo, jobID)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.baseURL+url, nil)
@@ -165,7 +184,7 @@ func (a *GiteaAPI) JobLogs(ctx context.Context, repo string, jobID int64) (strin
 		return "", err
 	}
 	if resp.StatusCode >= 300 {
-		return "", fmt.Errorf("GET %s: %d: %s", url, resp.StatusCode, body)
+		return "", statusError{method: http.MethodGet, path: url, code: resp.StatusCode, body: string(body)}
 	}
 	return string(body), nil
 }

e2e/host_test.go

@@ -0,0 +1,33 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import (
+	"os"
+	"testing"
+)
+
+func testHostMode(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	api, repo := newScenario(t)
+	hostname, err := os.Hostname()
+	if err != nil {
+		t.Fatalf("hostname: %v", err)
+	}
+	if err := api.CreateVariable(ctx, repo, "E2E_HOSTNAME", hostname); err != nil {
+		t.Fatalf("create variable: %v", err)
+	}
+	startRunner(t, repo, "e2e-host", runnerOptions{host: true})
+	pushWorkflow(t, api, repo, "host.yml")
+
+	wfRun := waitForRun(t, api, repo)
+	requireSuccess(t, api, repo, wfRun.ID)
+	if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-host-job-ran") == "" {
+		t.Errorf("job logs are missing the marker:\n%s", logs)
+	}
+}

e2e/main_test.go

@@ -6,11 +6,17 @@
 package e2e
 
 import (
+	"cmp"
 	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"os"
+	"os/exec"
+	"path/filepath"
 	"regexp"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -25,6 +31,16 @@ var fixture *GiteaFixture
 
 var skipReason string
 
+var binDir string
+
+var runnerBinary = sync.OnceValues(func() (string, error) {
+	bin := filepath.Join(binDir, "gitea-runner")
+	if out, err := exec.Command(cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil {
+		return "", fmt.Errorf("build runner: %w\n%s", err, out)
+	}
+	return bin, nil
+})
+
 func TestMain(m *testing.M) {
 	os.Exit(runSuite(m))
 }
@@ -47,6 +63,12 @@ func runSuite(m *testing.M) int {
 	fixture = f
 	defer func() { _ = fixture.Close(context.Background()) }()
 
+	if binDir, err = os.MkdirTemp("", "gitea-runner-e2e-"); err != nil {
+		fmt.Fprintf(os.Stderr, "create binary dir: %v\n", err)
+		return 1
+	}
+	defer os.RemoveAll(binDir)
+
 	fmt.Fprintf(os.Stderr, "gitea fixture: image=%s version=%s\n", fixture.image, fixture.version)
 
 	return m.Run()
@@ -109,6 +131,10 @@ func runLogs(t *testing.T, api *GiteaAPI, repo string, runID int64) string {
 	var all strings.Builder
 	for _, job := range jobs {
 		logs, err := api.JobLogs(ctx, repo, job.ID)
+		var response statusError
+		if errors.As(err, &response) && response.code == http.StatusNotFound {
+			continue // skipped jobs and reusable workflow callers never ran a task, so have no log
+		}
 		if err != nil {
 			t.Fatalf("job logs for %s: %v", job.Name, err)
 		}

e2e/reconnect_test.go

@@ -0,0 +1,127 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"net/url"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// flakyProxy forwards TCP to Gitea and can drop every connection to simulate an outage.
+type flakyProxy struct {
+	ln      net.Listener
+	target  string
+	down    atomic.Bool
+	refused atomic.Int64
+	mu      sync.Mutex
+	conns   map[net.Conn]struct{}
+}
+
+func newFlakyProxy(t *testing.T, targetURL string) *flakyProxy {
+	t.Helper()
+	u, err := url.Parse(targetURL)
+	if err != nil {
+		t.Fatalf("parse %q: %v", targetURL, err)
+	}
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	p := &flakyProxy{ln: ln, target: u.Host, conns: map[net.Conn]struct{}{}}
+	go p.serve()
+	t.Cleanup(func() {
+		_ = ln.Close()
+		p.dropAll()
+	})
+	return p
+}
+
+func (p *flakyProxy) URL() string { return "http://" + p.ln.Addr().String() }
+
+func (p *flakyProxy) serve() {
+	for {
+		c, err := p.ln.Accept()
+		if err != nil {
+			return
+		}
+		if p.down.Load() {
+			p.refused.Add(1)
+			_ = c.Close()
+			continue
+		}
+		up, err := net.Dial("tcp", p.target)
+		if err != nil {
+			_ = c.Close()
+			continue
+		}
+		p.mu.Lock()
+		p.conns[c], p.conns[up] = struct{}{}, struct{}{}
+		p.mu.Unlock()
+		go func() { _, _ = io.Copy(up, c); _ = up.Close() }()
+		go func() { _, _ = io.Copy(c, up); _ = c.Close() }()
+	}
+}
+
+func (p *flakyProxy) dropAll() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for c := range p.conns {
+		_ = c.Close()
+	}
+	clear(p.conns)
+}
+
+// cut starts the outage and returns once the runner has been refused at least refusals more times.
+func (p *flakyProxy) cut(t *testing.T, refusals int64) {
+	t.Helper()
+	before := p.refused.Load()
+	p.down.Store(true)
+	p.dropAll()
+	deadline := time.Now().Add(30 * time.Second)
+	for p.refused.Load() < before+refusals {
+		if time.Now().After(deadline) {
+			t.Fatal("runner never retried during the outage")
+		}
+		time.Sleep(pollInterval)
+	}
+}
+
+func (p *flakyProxy) restore() { p.down.Store(false) }
+
+// testReconnect checks that a runner recovers from a lost connection to Gitea, idle or mid-job.
+func testReconnect(t *testing.T) {
+	t.Parallel()
+	api, repo := newScenario(t)
+	proxy := newFlakyProxy(t, fixture.baseURL)
+	startRunner(t, repo, "e2e-reconnect", runnerOptions{host: true, address: proxy.URL()})
+
+	proxy.cut(t, 2)
+	pushWorkflow(t, api, repo, "reconnect_idle.yml")
+	idle := waitForRun(t, api, repo)
+	proxy.restore()
+	requireSuccess(t, api, repo, idle.ID)
+
+	pushWorkflow(t, api, repo, "reconnect.yml")
+	slow := waitForRunAfter(t, api, repo, idle.ID)
+	waitForRunningJobLog(t, api, repo, slow.ID, "e2e-before-outage")
+	// the job logs a line per second, so a long outage spans several that can only arrive after it
+	proxy.cut(t, 8)
+	proxy.restore()
+	requireSuccess(t, api, repo, slow.ID)
+	logs := runLogs(t, api, repo, slow.ID)
+	for i := 1; i <= 15; i++ {
+		if tick := fmt.Sprintf("tick-%02d", i); commandRow(logs, tick) != tick {
+			t.Errorf("log line %s written during or after the outage was lost:\n%s", tick, logs)
+			break
+		}
+	}
+}

e2e/reusable_test.go

@@ -0,0 +1,22 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import "testing"
+
+// testReusableWorkflow runs a local workflow_call workflow and consumes its output.
+func testReusableWorkflow(t *testing.T) {
+	t.Parallel()
+	api, repo := newScenario(t)
+	pushWorkflow(t, api, repo, "reusable_callee.yml")
+	pushWorkflow(t, api, repo, "reusable_caller.yml")
+
+	wfRun := waitForRun(t, api, repo)
+	requireSuccess(t, api, repo, wfRun.ID)
+	if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-reusable-ran") == "" {
+		t.Errorf("the called workflow did not run:\n%s", logs)
+	}
+}

e2e/runner_harness.go

@@ -30,11 +30,15 @@ type runnerOptions struct {
 	cacheHost  string
 	image      string
 	kubernetes *config.Kubernetes
+	host       bool   // register a host-mode label instead of a docker one
+	name       string // suffix that keeps runner names unique when a test starts several
+	address    string // Gitea address the runner talks to, default the fixture
 }
 
 func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *poll.Poller {
 	t.Helper()
 	ctx := t.Context()
+	address := cmp.Or(options.address, fixture.baseURL)
 	token, err := fixture.RegistrationToken(ctx, repo)
 	if err != nil {
 		t.Fatalf("get registration token: %v", err)
@@ -65,19 +69,23 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p
 		scheme, cfg.Kubernetes = labels.SchemeKubernetes, *options.kubernetes
 	}
 	rawLabel := labelName + ":" + scheme + "://" + cmp.Or(options.image, os.Getenv("E2E_JOB_IMAGE"))
+	if options.host {
+		rawLabel = labelName + ":" + labels.SchemeHost
+		cfg.Host.WorkdirParent = t.TempDir()
+	}
 	label, err := labels.Parse(rawLabel)
 	if err != nil {
 		t.Fatalf("parse label %q: %v", rawLabel, err)
 	}
 	labelNames := []string{label.Name}
 
-	pingCli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders)
+	pingCli := client.New(address, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders)
-	if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name()}); err != nil {
+	if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name() + options.name}); err != nil {
-		t.Fatalf("ping %s: %v", fixture.baseURL, err)
+		t.Fatalf("ping %s: %v", address, err)
 	}
 
 	regResp, err := pingCli.Register(ctx, &runnerv1.RegisterRequest{
-		Name:         t.Name(),
+		Name:         t.Name() + options.name,
 		Token:        token,
 		Version:      "e2e",
 		Labels:       labelNames,
@@ -96,12 +104,12 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p
 		UUID:      regResp.Runner.Uuid,
 		Name:      regResp.Runner.Name,
 		Token:     regResp.Runner.Token,
-		Address:   fixture.baseURL,
+		Address:   address,
 		Labels:    []string{rawLabel},
 		Ephemeral: regResp.Runner.Ephemeral,
 	}
 
-	cli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), reg.UUID, reg.Token, config.RequestTimeout, cfg.Runner.ExtraHeaders)
+	cli := client.New(address, cfg.Runner.TLSConfig(), reg.UUID, reg.Token, config.RequestTimeout, cfg.Runner.ExtraHeaders)
 
 	runner := run.NewRunner(cfg, reg, cli)
 	if _, err := runner.Declare(ctx, labelNames); err != nil {

e2e/scheduling_test.go

@@ -0,0 +1,113 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+//go:build e2e
+
+package e2e
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// testLabelRouting checks that each job reaches the runner owning its label and that a
+// job with no matching runner stays queued.
+func testLabelRouting(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	api, repo := newScenario(t)
+	startRunner(t, repo, "e2e-route-a", runnerOptions{host: true, name: "-a"})
+	startRunner(t, repo, "e2e-route-b", runnerOptions{host: true, name: "-b"})
+	pushWorkflow(t, api, repo, "route.yml")
+
+	wfRun := waitForRun(t, api, repo)
+	ctx, cancel := context.WithTimeout(ctx, runTimeout)
+	defer cancel()
+	var jobs map[string]ActionJob
+	for {
+		list, err := api.Jobs(ctx, repo, wfRun.ID)
+		if err != nil {
+			t.Fatalf("list jobs: %v", err)
+		}
+		jobs = map[string]ActionJob{}
+		for _, job := range list {
+			jobs[job.Name] = job
+		}
+		if jobs["a"].Status == "completed" && jobs["b"].Status == "completed" {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			t.Fatalf("jobs a and b did not complete: %+v", list)
+		case <-time.After(pollInterval):
+		}
+	}
+
+	for name, suffix := range map[string]string{"a": "-a", "b": "-b"} {
+		if got := jobs[name].RunnerName; !strings.HasSuffix(got, suffix) {
+			t.Errorf("job %s ran on runner %q, want the one named *%s", name, got, suffix)
+		}
+		if jobs[name].Conclusion != "success" {
+			t.Errorf("job %s concluded %q", name, jobs[name].Conclusion)
+		}
+	}
+	// a and b were picked up after unmatched was queued, so a runner that could take it would have by now
+	if status := jobs["unmatched"].Status; status == "in_progress" || status == "completed" {
+		t.Errorf("job with no matching runner is %q, want it queued", status)
+	}
+}
+
+// testCapacity checks that runner.capacity bounds how many jobs run at once.
+func testCapacity(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name       string
+		capacity   int
+		concurrent bool
+	}{
+		{"serial", 1, false},
+		{"parallel", 2, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			api, repo := newScenario(t)
+			startRunner(t, repo, "e2e-capacity", runnerOptions{capacity: tc.capacity})
+			pushWorkflow(t, api, repo, "capacity.yml")
+
+			wfRun := waitForRun(t, api, repo)
+			requireSuccess(t, api, repo, wfRun.ID)
+
+			jobs, err := api.Jobs(t.Context(), repo, wfRun.ID)
+			if err != nil || len(jobs) != 2 {
+				t.Fatalf("list jobs: %v (%d jobs)", err, len(jobs))
+			}
+			var start, end [2]int64
+			for i, job := range jobs {
+				logs, err := api.JobLogs(t.Context(), repo, job.ID)
+				if err != nil {
+					t.Fatalf("job logs: %v", err)
+				}
+				start[i] = spanValue(t, logs, "span-start=")
+				end[i] = spanValue(t, logs, "span-end=")
+			}
+			overlap := max(start[0], start[1]) < min(end[0], end[1])
+			if overlap != tc.concurrent {
+				t.Errorf("capacity %d: jobs overlapped=%v, want %v (starts %v ends %v)", tc.capacity, overlap, tc.concurrent, start, end)
+			}
+		})
+	}
+}
+
+func spanValue(t *testing.T, logs, prefix string) int64 {
+	t.Helper()
+	row := commandRow(logs, prefix)
+	n, err := strconv.ParseInt(strings.TrimPrefix(row, prefix), 10, 64)
+	if err != nil {
+		t.Fatalf("no %q value in logs:\n%s", prefix, logs)
+	}
+	return n
+}

e2e/testdata/workflows/capacity.yml

@@ -0,0 +1,17 @@
+name: capacity
+on: push
+jobs:
+  one:
+    runs-on: e2e-capacity
+    steps:
+      - run: |
+          echo "span-start=$(date +%s%N)"
+          sleep 8
+          echo "span-end=$(date +%s%N)"
+  two:
+    runs-on: e2e-capacity
+    steps:
+      - run: |
+          echo "span-start=$(date +%s%N)"
+          sleep 8
+          echo "span-end=$(date +%s%N)"

e2e/testdata/workflows/cli.yml

@@ -0,0 +1,7 @@
+name: cli
+on: push
+jobs:
+  hello:
+    runs-on: e2e-cli
+    steps:
+      - run: echo e2e-cli-job-ran

e2e/testdata/workflows/cli_labels.yml

@@ -0,0 +1,7 @@
+name: cli-labels
+on: push
+jobs:
+  hello:
+    runs-on: e2e-override
+    steps:
+      - run: echo e2e-override-ran

e2e/testdata/workflows/cli_once.yml

@@ -0,0 +1,11 @@
+name: cli-once
+on: push
+jobs:
+  first:
+    runs-on: e2e-once
+    steps:
+      - run: echo e2e-once-first
+  second:
+    runs-on: e2e-once
+    steps:
+      - run: echo e2e-once-second

e2e/testdata/workflows/cli_shutdown.yml

@@ -0,0 +1,10 @@
+name: cli-shutdown
+on: push
+jobs:
+  slow:
+    runs-on: e2e-shutdown
+    steps:
+      - run: |
+          echo e2e-shutdown-started
+          sleep 8
+          echo e2e-shutdown-finished

e2e/testdata/workflows/exec_artifact.yml

@@ -0,0 +1,15 @@
+on: push
+jobs:
+  roundtrip:
+    runs-on: ubuntu-latest
+    steps:
+      - run: mkdir -p out && echo exec-payload > out/data.txt
+      - uses: actions/upload-artifact@v7
+        with:
+          name: exec-artifact
+          path: out/data.txt
+      - uses: actions/download-artifact@v8
+        with:
+          name: exec-artifact
+          path: got
+      - run: grep -q exec-payload got/data.txt

e2e/testdata/workflows/failure.yml

@@ -0,0 +1,30 @@
+name: failure
+on: push
+jobs:
+  fail:
+    runs-on: ubuntu-latest
+    steps:
+      - run: echo e2e-before-fail
+      - run: exit 3
+      - if: success()
+        run: echo e2e-skipped-step
+      - if: failure()
+        run: echo e2e-failure-cleanup
+  tolerated:
+    runs-on: ubuntu-latest
+    steps:
+      - continue-on-error: true
+        run: exit 1
+      - run: echo e2e-tolerated-next
+  dependent:
+    needs: fail
+    runs-on: ubuntu-latest
+    steps:
+      - run: echo dependent
+  timeout:
+    runs-on: ubuntu-latest
+    timeout-minutes: 1
+    steps:
+      - run: |
+          echo e2e-timeout-started
+          sleep 300

e2e/testdata/workflows/host.yml

@@ -0,0 +1,10 @@
+name: host
+on: push
+jobs:
+  host:
+    runs-on: e2e-host
+    steps:
+      - run: |
+          test "$(hostname)" = "${{ vars.E2E_HOSTNAME }}"
+          test -n "$GITHUB_WORKSPACE"
+          echo e2e-host-job-ran

e2e/testdata/workflows/reconnect.yml

@@ -0,0 +1,15 @@
+name: reconnect
+on:
+  push:
+    paths:
+      - .gitea/workflows/reconnect.yml
+jobs:
+  slow:
+    runs-on: e2e-reconnect
+    steps:
+      - run: |
+          echo e2e-before-outage
+          for i in $(seq -w 1 15); do
+            echo "tick-$i"
+            sleep 1
+          done

e2e/testdata/workflows/reconnect_idle.yml

@@ -0,0 +1,10 @@
+name: reconnect-idle
+on:
+  push:
+    paths:
+      - .gitea/workflows/reconnect_idle.yml
+jobs:
+  hello:
+    runs-on: e2e-reconnect
+    steps:
+      - run: echo e2e-idle-recovered