enhance(kubernetes): gzip copies into job containers over exec (#1355)

The Kubernetes backend copies actions and other files into the job container with `tar` over `pods/exec`, so the data goes through the API server, which on managed clusters often sits in other zones than the nodes (each copied byte then crosses zones twice). Actions are megabytes of JavaScript bundles that gzip about fivefold:

| Action | tar | gzip |
|---|---|---|
| actions/setup-go@v5 | 8.0 MB | 1.5 MB |
| actions/cache@v4 | 11.6 MB | 2.0 MB |
| actions/setup-java@v4 | 15.3 MB | 2.5 MB |
| docker/build-push-action@v6 | 12.9 MB | 3.0 MB |

The startup probe now also reports whether the job container has `gzip`. If it does, `CopyDir` and copies of 64 KiB or more are gzipped at `BestSpeed` and extracted with `gzip -dc | tar -xf -`. Smaller copies (run scripts, file command files) stay uncompressed, where gzip would save next to nothing and cost a process in the pod. Images without `gzip` keep the plain tar stream, so the image requirements do not change.

This is independent of #1354 and gets most of its traffic saving with no new moving parts.

---
*Written by Codet (AI coding agent) on behalf of @lunny.*

---------

Co-authored-by: Zettat123 <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1355
Reviewed-by: Zettat123 <[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/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")
-		c.pod.arch, c.pod.path = goArchToActionArch(machine), path
+		probed := strings.SplitN(output, "\n", 3)
+		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_composite.go

@@ -110,6 +110,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/expression.go

@@ -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/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

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)
-	if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name()}); err != nil {
-		t.Fatalf("ping %s: %v", fixture.baseURL, err)
+	pingCli := client.New(address, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders)
+	if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name() + options.name}); err != nil {
+		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

e2e/testdata/workflows/reusable_callee.yml

@@ -0,0 +1,20 @@
+name: reusable-callee
+on:
+  workflow_call:
+    inputs:
+      who:
+        type: string
+        required: true
+    outputs:
+      greeting:
+        value: ${{ jobs.say.outputs.greeting }}
+jobs:
+  say:
+    runs-on: ubuntu-latest
+    outputs:
+      greeting: ${{ steps.greet.outputs.greeting }}
+    steps:
+      - id: greet
+        run: |
+          echo "greeting=hello-${{ inputs.who }}" >> "$GITHUB_OUTPUT"
+          echo e2e-reusable-ran

e2e/testdata/workflows/reusable_caller.yml

@@ -0,0 +1,12 @@
+name: reusable-caller
+on: push
+jobs:
+  call:
+    uses: ./.gitea/workflows/reusable_callee.yml
+    with:
+      who: e2e
+  verify:
+    needs: call
+    runs-on: ubuntu-latest
+    steps:
+      - run: test "${{ needs.call.outputs.greeting }}" = "hello-e2e"

e2e/testdata/workflows/route.yml

@@ -0,0 +1,15 @@
+name: route
+on: push
+jobs:
+  a:
+    runs-on: e2e-route-a
+    steps:
+      - run: echo route-a
+  b:
+    runs-on: e2e-route-b
+    steps:
+      - run: echo route-b
+  unmatched:
+    runs-on: e2e-route-none
+    steps:
+      - run: echo route-none

internal/app/cmd/exec.go

@@ -7,12 +7,11 @@ package cmd
 import (
 	"cmp"
 	"context"
-	"crypto/rand"
-	"encoding/hex"
 	"encoding/json/v2"
 	"errors"
 	"fmt"
 	"maps"
+	"net"
 	"os"
 	"path/filepath"
 	"slices"
@@ -481,24 +480,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
 			execArgs.artifactServerPath = tempDir
 		}
 
-		// Register ACTIONS_RUNTIME_TOKEN against local cache server
 		env := execArgs.LoadEnvs()
-		const actionsRuntimeTokenEnvName = "ACTIONS_RUNTIME_TOKEN"
-		actionsRuntimeToken := env[actionsRuntimeTokenEnvName]
-		if actionsRuntimeToken == "" {
-			actionsRuntimeToken = os.Getenv(actionsRuntimeTokenEnvName)
-		}
-		if actionsRuntimeToken == "" {
-			tmpBranch := make([]byte, 12)
-			if _, err := rand.Read(tmpBranch); err != nil {
-				actionsRuntimeToken = "token"
-			} else {
-				actionsRuntimeToken = hex.EncodeToString(tmpBranch)
-			}
-			env[actionsRuntimeTokenEnvName] = actionsRuntimeToken
-			os.Setenv(actionsRuntimeTokenEnvName, actionsRuntimeToken)
-		}
-		handler.RegisterJob(actionsRuntimeToken, artifactcache.JobCredential{Repo: "__local/__exec"})
 
 		// no service aliases: exec builds one config for the whole plan
 		run.BypassProxyForDockerHost(os.Getenv("DOCKER_HOST"))
@@ -517,7 +499,22 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
 		}
 
 		config.Env["ACT_EXEC"] = "true"
-		config.Secrets[actionsRuntimeTokenEnvName] = actionsRuntimeToken
+		inheritedToken := cmp.Or(env["ACTIONS_RUNTIME_TOKEN"], os.Getenv("ACTIONS_RUNTIME_TOKEN"))
+		config.RuntimeTokens = map[*model.Workflow]string{}
+		sameFile := map[string]int{}
+		for _, stage := range plan.Stages {
+			for _, run := range stage.Runs {
+				if config.RuntimeTokens[run.Workflow] != "" {
+					continue
+				}
+				sameFile[run.Workflow.File]++
+				workflowRun := fmt.Sprintf("%s/%s/%d", cmp.Or(env["GITHUB_RUN_ID"], "1"), run.Workflow.File, sameFile[run.Workflow.File]) // each workflow its own run as on GitHub, also when file names repeat across directories
+				token := cmp.Or(inheritedToken, artifacts.LocalRuntimeToken(workflowRun, maxLifetime))
+				config.RuntimeTokens[run.Workflow] = token
+				config.ExtraMasks = append(config.ExtraMasks, token)
+				handler.RegisterJob(token, artifactcache.JobCredential{Repo: "__local/__exec", Results: "http://" + net.JoinHostPort(execArgs.artifactServerAddr, execArgs.artifactServerPort)})
+			}
+		}
 
 		config.Token = cmp.Or(config.Secrets["GITEA_TOKEN"], config.Secrets["GITHUB_TOKEN"])
 
@@ -580,9 +577,9 @@ func loadExecCmd(ctx context.Context) *cobra.Command {
 	execCmd.Flags().StringArrayVarP(&execArg.containerCapAdd, "container-cap-add", "", []string{}, "kernel capabilities to add to the workflow containers (e.g. --container-cap-add SYS_PTRACE)")
 	execCmd.Flags().StringArrayVarP(&execArg.containerCapDrop, "container-cap-drop", "", []string{}, "kernel capabilities to remove from the workflow containers (e.g. --container-cap-drop SYS_PTRACE)")
 	execCmd.Flags().StringVarP(&execArg.containerOptions, "container-opts", "", "", "container options")
-	execCmd.PersistentFlags().StringVarP(&execArg.artifactServerPath, "artifact-server-path", "", ".", "Defines the path where the artifact server stores uploads and retrieves downloads from. If not specified the artifact server will not start.")
+	execCmd.PersistentFlags().StringVarP(&execArg.artifactServerPath, "artifact-server-path", "", "", "Defines the path where the artifact server stores uploads and retrieves downloads from, outside the workspace that jobs copy. Defaults to a temporary directory removed on exit.")
 	execCmd.PersistentFlags().StringVarP(&execArg.artifactServerAddr, "artifact-server-addr", "", "", "Defines the address where the artifact server listens")
-	execCmd.PersistentFlags().StringVarP(&execArg.artifactServerPort, "artifact-server-port", "", "34567", "Defines the port where the artifact server listens (will only bind to localhost).")
+	execCmd.PersistentFlags().StringVarP(&execArg.artifactServerPort, "artifact-server-port", "", "34567", "Defines the port where the artifact server listens")
 	execCmd.PersistentFlags().StringVarP(&execArg.defaultActionsURL, "default-actions-url", "", "https://github.com", "Defines the default url of action instance.")
 	execCmd.PersistentFlags().BoolVarP(&execArg.noSkipCheckout, "no-skip-checkout", "", false, "Do not skip actions/checkout")
 	execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log")