runner/internal/app/run/runner.go

// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package run

import (
	"bytes"
	"cmp"
	"context"
	"encoding/base64"
	"encoding/json/v2"
	"errors"
	"fmt"
	"maps"
	"net"
	"net/http"
	"net/netip"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"gitea.com/gitea/runner/act/artifactcache"
	"gitea.com/gitea/runner/act/common"
	"gitea.com/gitea/runner/act/common/logging"
	"gitea.com/gitea/runner/act/container"
	"gitea.com/gitea/runner/act/runner"
	"gitea.com/gitea/runner/internal/pkg/client"
	"gitea.com/gitea/runner/internal/pkg/config"
	"gitea.com/gitea/runner/internal/pkg/disk"
	"gitea.com/gitea/runner/internal/pkg/envcheck"
	"gitea.com/gitea/runner/internal/pkg/labels"
	"gitea.com/gitea/runner/internal/pkg/metrics"
	"gitea.com/gitea/runner/internal/pkg/report"
	"gitea.com/gitea/runner/internal/pkg/telemetry"
	"gitea.com/gitea/runner/internal/pkg/ver"

	"connectrpc.com/connect/v2"
	"gitea.dev/actionslib/pkg/model"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	docker_container "github.com/moby/moby/api/types/container"
	"go.yaml.in/yaml/v4"
)

// CapabilityCancelling tells the server this runner understands the
// transitional cancelling state and will run post-step cleanup before
// finalizing a task as RESULT_CANCELLED.
const CapabilityCancelling = "cancelling"

// RunnerCapabilities are the capability flags this runner advertises to the
// server during registration and declaration. The server uses them to enable
// transitional features that require runner-side support.
func RunnerCapabilities() []string {
	return []string{CapabilityCancelling}
}

// Runner runs the pipeline.
type Runner struct {
	name string
	uuid string

	cfg *config.Config

	client       client.Client
	labels       labels.Labels
	envs         map[string]string
	cacheHandler *artifactcache.Handler
	capabilities string

	isolatedCacheContainer func() string
	instanceTransport      http.RoundTripper

	runningTasks            sync.Map
	runningCount            atomic.Int64
	lastIdleCleanupUnixNano atomic.Int64
	now                     func() time.Time
	healthCheckLast         time.Time
	healthCheckReady        bool
	healthCheckReason       string
	healthStatusSet         bool
	healthStatusReady       bool
	healthStatusReason      string
	runHealthCheck          func(context.Context, string, time.Duration, map[string]string) error
}

func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) *Runner {
	ls := labels.Labels{}
	for _, v := range reg.Labels {
		if l, err := labels.Parse(v); err == nil {
			ls = append(ls, l)
		}
	}
	envs := make(map[string]string, len(cfg.Runner.Envs))
	maps.Copy(envs, cfg.Runner.Envs)
	var cacheHandler *artifactcache.Handler
	if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
		warnIgnoredCachePolicy(cfg)
		warnIgnoredCacheSecret(cfg)
		handler, err := artifactcache.StartHandler(artifactcache.Options{
			Dir:        cfg.Cache.Dir,
			OutboundIP: cfg.Cache.Host,
			Port:       cfg.Cache.Port,
			Upstream:   cfg.Cache.ExternalServer,
			S3:         cacheS3(cfg),
			Policy:     CachePolicy(cfg),
			Logger:     logging.Default().With("module", "cache_request"),
		})
		if err != nil {
			logging.Errorf("cannot init cache server, it will be disabled: %v", err)
			// go on
		} else {
			cacheHandler = handler
			envs["ACTIONS_CACHE_URL"] = handler.ExternalURL() + "/"
		}
	}

	// set artifact gitea api
	envs["ACTIONS_RUNTIME_URL"] = strings.TrimSuffix(cli.Address(), "/") + "/api/actions_pipeline/"
	envs["ACTIONS_RESULTS_URL"] = strings.TrimSuffix(cli.Address(), "/")

	// Set specific environments to distinguish between Gitea and GitHub
	envs["GITEA_ACTIONS"] = "true"
	envs["GITEA_ACTIONS_RUNNER_VERSION"] = ver.Version()

	runner := &Runner{
		name:           reg.Name,
		uuid:           reg.UUID,
		cfg:            cfg,
		client:         cli,
		labels:         ls,
		envs:           envs,
		cacheHandler:   cacheHandler,
		now:            time.Now,
		runHealthCheck: executeHealthCheck,
	}
	runner.isolatedCacheContainer = sync.OnceValue(runner.detectIsolatedCacheContainer)
	runner.instanceTransport = client.NewTransport(cli.Address(), cfg.Runner.TLSConfig())
	return runner
}

// Close shuts down the cache server this runner exposes to job containers.
func (r *Runner) Close() error {
	return r.cacheHandler.Close()
}

// removeOrphanNetworks is a variable so tests can substitute one that needs no Docker daemon.
var removeOrphanNetworks = container.RemoveOrphanNetworks

var removeOrphanJobVolumes = container.RemoveOrphanJobVolumes

var removeOrphanKubernetesResources = container.RemoveOrphanKubernetesResources

const orphanDockerResourceAge = 10 * time.Minute // spares the network and volumes of a job still starting its first container

// OnIdle performs lightweight maintenance during polling idle windows.
// It runs synchronously on the poller goroutine; shouldRunIdleCleanup
// throttles invocations to runner.idle_cleanup_interval so the impact on
// poll cadence is bounded even when the workdir root is large.
func (r *Runner) OnIdle(ctx context.Context) {
	if !r.shouldRunIdleCleanup() {
		return
	}
	// Bind-workdir mode: reclaim stale per-task workspace dirs (numeric task IDs).
	if r.cfg.Container.BindWorkdir {
		r.cleanupStaleDirs(ctx, containerWorkdirRoot(r.cfg.Container.WorkdirParent), isTaskIDDir)
	}
	// Host mode: reclaim per-job scratch dirs left behind when HostEnvironment
	// cleanup timed out (e.g. a delete stalled by an AV/EDR filter driver). They
	// sit under the host workdir parent next to a shared tool_cache, which the name
	// match leaves untouched. No-op when no host-mode job ever ran.
	if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" {
		r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir)
	}
	r.cleanupOrphanDockerResources(ctx)
	if r.uuid != "" && r.usesKubernetes() {
		ctx, cancel := context.WithTimeout(ctx, time.Minute)
		defer cancel()
		if err := removeOrphanKubernetesResources(ctx, r.kubernetesOptions(nil), r.now().Add(-r.cfg.Runner.WorkdirCleanupAge)); err != nil {
			logging.Warnf("failed to clean up job pods left behind by earlier jobs: %v", err)
		}
	}
}

func (r *Runner) cleanupOrphanDockerResources(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()
	if r.uuid == "" || (!r.requiresDocker() && !dockerReachable(ctx)) {
		return
	}
	cutoff := r.now().Add(-orphanDockerResourceAge)
	if err := removeOrphanNetworks(ctx, r.uuid, r.isolatedCacheContainer(), cutoff); err != nil {
		logging.Warnf("failed to clean up networks left behind by earlier jobs: %v", err)
	}
	if err := removeOrphanJobVolumes(ctx, r.uuid, cutoff); err != nil {
		logging.Warnf("failed to clean up volumes left behind by earlier jobs: %v", err)
	}
}

func (r *Runner) shouldRunIdleCleanup() bool {
	if r.cfg.Runner.WorkdirCleanupAge <= 0 || r.cfg.Runner.IdleCleanupInterval <= 0 {
		return false
	}
	if r.RunningCount() != 0 {
		return false
	}
	now := r.now()
	interval := r.cfg.Runner.IdleCleanupInterval
	for {
		last := r.lastIdleCleanupUnixNano.Load()
		if last != 0 && now.Sub(time.Unix(0, last)) < interval {
			return false
		}
		if r.lastIdleCleanupUnixNano.CompareAndSwap(last, now.UnixNano()) {
			return true
		}
	}
}

// isTaskIDDir reports whether name is a per-task workspace dir (numeric task
// ID). Any other directory is skipped to avoid deleting operator-managed data
// under workdir_root.
func isTaskIDDir(name string) bool {
	_, err := strconv.ParseUint(name, 10, 64)
	return err == nil
}

// isHostScratchDir reports whether name is a per-job host-mode scratch dir:
// hex.EncodeToString of 8 random bytes, i.e. exactly 16 lowercase hex chars
// (see startHostEnvironment in act/runner/run_context.go). The narrow match
// leaves a sibling shared "tool_cache" dir and any operator data untouched.
func isHostScratchDir(name string) bool {
	if len(name) != 16 {
		return false
	}
	for _, c := range name {
		if (c < '0' || c > '9') && (c < 'a' || c > 'f') {
			return false
		}
	}
	return true
}

// cleanupStaleDirs removes immediate child directories of root that match and
// whose mtime is older than WorkdirCleanupAge. It is a no-op when root does not
// exist yet (the runner has never written there).
func (r *Runner) cleanupStaleDirs(ctx context.Context, root string, match func(name string) bool) {
	entries, err := os.ReadDir(root)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return
		}
		logging.Warnf("failed to list directory %s for stale cleanup: %v", root, err)
		return
	}

	// A task may begin between shouldRunIdleCleanup's running-count check and
	// the loop below. That is safe because new dirs are created with the
	// current mtime and therefore fall on the keep side of cutoff.
	cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge)
	for _, entry := range entries {
		if err := ctx.Err(); err != nil {
			return
		}
		if !entry.IsDir() || !match(entry.Name()) {
			continue
		}
		info, err := entry.Info()
		if err != nil {
			logging.Warnf("failed to stat %s: %v", filepath.Join(root, entry.Name()), err)
			continue
		}
		if info.ModTime().After(cutoff) {
			continue
		}
		dir := filepath.Join(root, entry.Name())
		if err := os.RemoveAll(dir); err != nil {
			logging.Warnf("failed to clean stale directory %s: %v", dir, err)
			continue
		}
		logging.Infof("cleaned stale directory %s", dir)
	}
}

func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) (err error) {
	if _, ok := r.runningTasks.LoadOrStore(task.Id, struct{}{}); ok {
		return fmt.Errorf("task %d is already running", task.Id)
	}
	defer r.runningTasks.Delete(task.Id)

	r.runningCount.Add(1)
	defer r.runningCount.Add(-1)

	start := time.Now()

	ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
	defer cancel()
	ctx, endJob := telemetry.StartJob(ctx, task)
	// A proxy URL may carry credentials, and every job is given it; keep them out of the log.
	reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg, proxyPasswords(r.envs)...)
	var volumeCleanup []common.Executor
	var volumeCleanupMu sync.Mutex
	if r.cfg.Runner.PostTaskScript == "" {
		ctx = runner.WithJobVolumeCleanup(ctx, func(cleanup common.Executor) {
			volumeCleanupMu.Lock()
			defer volumeCleanupMu.Unlock()
			volumeCleanup = append(volumeCleanup, cleanup)
		})
	}
	var runErr error
	defer func() {
		lastWords := ""
		if runErr != nil {
			lastWords = runErr.Error()
		}
		err = reporter.Close(lastWords)
		if err := cleanupJobVolumes(ctx, volumeCleanup); err != nil {
			logging.Warnf("task %d volume cleanup after reporting: %v", task.Id, err)
		}

		metrics.JobDuration.Observe(time.Since(start).Seconds())
		metrics.JobsTotal.WithLabelValues(metrics.JobStatus(ctx, reporter.Result())).Inc()
		endJob(reporter.Result())
	}()
	reporter.RunDaemon()
	runErr = r.run(ctx, task, reporter)

	return nil
}

func cleanupJobVolumes(ctx context.Context, cleanups []common.Executor) error {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Minute)
	defer cancel()
	var errs []error
	for _, cleanup := range cleanups {
		errs = append(errs, cleanup(ctx))
	}
	return errors.Join(errs...)
}

func (r *Runner) cloneEnvs() map[string]string {
	// Reserve space for the per-task keys injected by run():
	// ACTIONS_ID_TOKEN_REQUEST_URL, ACTIONS_ID_TOKEN_REQUEST_TOKEN, ACTIONS_RUNTIME_TOKEN,
	// GITEA_ACTIONS_CAPABILITIES, GITEA_RUN_ID.
	envs := make(map[string]string, len(r.envs)+5)
	maps.Copy(envs, r.envs)
	return envs
}

// restoreCalledWorkflowContext undoes Gitea's rewrite of `event_name` and `event.inputs` for older runners,
// and returns the job's `inputs` from `gitea_workflow_call`, nil if it is not a called workflow's job.
// See https://github.com/go-gitea/gitea/pull/39452
func restoreCalledWorkflowContext(task *runnerv1.Task) map[string]any {
	workflowCall := task.GetContext().GetFields()["gitea_workflow_call"].GetStructValue().GetFields()
	if workflowCall == nil {
		return nil
	}
	task.Context.Fields["event_name"] = workflowCall["original_event_name"]
	if event := task.Context.Fields["event"].GetStructValue(); event != nil {
		if inputs, ok := workflowCall["original_event_inputs"]; ok {
			event.Fields["inputs"] = inputs
		} else {
			delete(event.Fields, "inputs")
		}
	}
	return workflowCall["inputs"].GetStructValue().AsMap()
}

// getDefaultActionsURL
// when DEFAULT_ACTIONS_URL == "https://github.com" and GithubMirror is not blank,
// it should be set to GithubMirror first.
func (r *Runner) getDefaultActionsURL(task *runnerv1.Task) string {
	giteaDefaultActionsURL := task.Context.Fields["gitea_default_actions_url"].GetStringValue()
	if giteaDefaultActionsURL == "https://github.com" && r.cfg.Runner.GithubMirror != "" {
		return r.cfg.Runner.GithubMirror
	}
	return giteaDefaultActionsURL
}

// isSelfHostedActionsURL reports whether actions resolve to this self-hosted Gitea
// (DEFAULT_ACTIONS_URL=self), i.e. gitea_default_actions_url is AppURL rather than
// github.com (which may be mirror-substituted by getDefaultActionsURL). Only then may the
// task token be attached to action clone URLs on the actions instance host.
func (r *Runner) isSelfHostedActionsURL(task *runnerv1.Task) bool {
	giteaDefaultActionsURL := task.Context.Fields["gitea_default_actions_url"].GetStringValue()
	return giteaDefaultActionsURL != "" && giteaDefaultActionsURL != "https://github.com"
}

// dockerReachable is a variable so tests can substitute one that needs no Docker daemon. It
// probes the environment act connects through, not container.docker_host, which act ignores.
var dockerReachable = func(ctx context.Context) bool {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	return envcheck.CheckIfDockerRunning(ctx, "") == nil
}

func (r *Runner) requiresDocker() bool {
	return r.labels.RequireDocker() || r.cfg.Container.RequireDocker
}

func (r *Runner) usesKubernetes() bool {
	return slices.ContainsFunc(r.labels, func(label *labels.Label) bool { return label.Schema == labels.SchemeKubernetes })
}

// kubernetesOptions layers the pod templates of the runner labels a job runs on, in runs-on order, over the shared one.
func (r *Runner) kubernetesOptions(runsOn []string) container.KubernetesOptions {
	options := container.KubernetesOptions{
		Kubeconfig:   r.cfg.Kubernetes.Kubeconfig,
		Namespace:    r.cfg.Kubernetes.Namespace,
		PodTemplates: []map[string]any{r.cfg.Kubernetes.PodTemplate},
		RunnerUUID:   r.uuid,
	}
	if transfer := r.cfg.Kubernetes.Transfer; transfer.Mode == config.KubernetesTransferModeAuto || transfer.Mode == config.KubernetesTransferModeDirect {
		options.Transfer = &container.KubernetesTransfer{
			Host: transfer.Host, Port: transfer.Port, Image: transfer.Image, Fallback: transfer.Mode == config.KubernetesTransferModeAuto,
		}
	}
	names := r.labels.Names()
	for index, name := range runsOn {
		if template, ok := r.cfg.Kubernetes.PodTemplates[name]; ok && slices.Contains(names, name) && !slices.Contains(runsOn[:index], name) {
			options.PodTemplates = append(options.PodTemplates, template)
		}
	}
	return options
}

// fallbackPlatform is where a job runs whose runs-on matches no label, as any job without a
// runs-on does, since Gitea sends those to every runner.
func (r *Runner) fallbackPlatform(ctx context.Context) string {
	switch {
	case r.requiresDocker():
		return r.cfg.Runner.DefaultImage
	case r.usesKubernetes():
		return labels.SchemeKubernetes + "://" + r.cfg.Runner.DefaultImage
	case dockerReachable(ctx):
		return r.cfg.Runner.DefaultImage
	}
	return labels.SelfHostedPlatform
}

func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.Reporter) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic: %v", r)
		}
	}()

	workflowCallInputs := restoreCalledWorkflowContext(task)
	for _, line := range r.setupLines(task) {
		reporter.Logf("%s", line)
	}

	workflow, jobID, err := generateWorkflow(task)
	if err != nil {
		return err
	}
	logging.With(taskLogFields(task, jobID)...).
		With("actions_url", r.getDefaultActionsURL(task), "address", r.client.Address()).
		Info("task started")

	plan, err := model.CombineWorkflowPlanner(workflow).PlanJob(jobID)
	if err != nil {
		return err
	}
	job := workflow.GetJob(jobID)
	reporter.ResetSteps(job.Steps)
	telemetry.SetJobName(ctx, cmp.Or(job.Name, jobID))

	taskContext := task.Context.Fields
	envs := r.cloneEnvs()

	// act asks for the platform once per step, so resolve the fallback at most once per task.
	fallbackPlatform := sync.OnceValue(func() string { return r.fallbackPlatform(ctx) })
	platformPicker := func(runsOn []string) string {
		if platform := r.labels.PickPlatform(runsOn); platform != "" {
			return platform
		}
		return fallbackPlatform()
	}
	platform := platformPicker(job.RunsOn())
	cacheURL, cacheContainer := r.cacheForJob(job, platform)
	if cacheURL != "" {
		envs["ACTIONS_CACHE_URL"] = cacheURL + "/"
	}

	// Added per task because this job's service containers must be reached directly, and
	// act reaches them by their workflow key.
	var services map[string]yaml.Node
	_ = job.RawServices.Decode(&services) // a whole-value expression names no service before evaluation
	serviceNames := slices.Sorted(maps.Keys(services))
	if strings.HasPrefix(platform, labels.SchemeKubernetes+"://") {
		var hostnames []string
		for _, name := range serviceNames {
			hostnames = append(hostnames, container.KubernetesServiceHostnames(name)...)
		}
		serviceNames = hostnames
	}
	proxyEnv := JobProxyEnv(envs, cacheURL, serviceNames)
	maps.Copy(envs, proxyEnv)

	if r.capabilities != "" {
		envs["GITEA_ACTIONS_CAPABILITIES"] = r.capabilities
	}
	if v := taskContext["run_id"].GetStringValue(); v != "" {
		envs["GITEA_RUN_ID"] = v
	}
	telemetry.AddJobEnv(ctx, envs)

	preset := model.GithubContextFromMap(task.Context.AsMap())
	preset.Token = cmp.Or(task.Secrets["GITEA_TOKEN"], task.Secrets["GITHUB_TOKEN"], preset.Token)

	if actionsIDTokenRequestURL := taskContext["actions_id_token_request_url"].GetStringValue(); actionsIDTokenRequestURL != "" {
		envs["ACTIONS_ID_TOKEN_REQUEST_URL"] = actionsIDTokenRequestURL
		envs["ACTIONS_ID_TOKEN_REQUEST_TOKEN"] = taskContext["actions_id_token_request_token"].GetStringValue()
		task.Secrets["ACTIONS_ID_TOKEN_REQUEST_TOKEN"] = envs["ACTIONS_ID_TOKEN_REQUEST_TOKEN"]
	}

	giteaRuntimeToken := taskContext["gitea_runtime_token"].GetStringValue()
	if giteaRuntimeToken == "" {
		// use task token to action api token for previous Gitea Server Versions
		giteaRuntimeToken = preset.Token
	}
	envs["ACTIONS_RUNTIME_TOKEN"] = giteaRuntimeToken
	// Mask the runtime token so it cannot be echoed in user step output; it is
	// now also the cache server's bearer credential and leaking it would let
	// any reader of the log impersonate this job against the cache.
	if giteaRuntimeToken != "" {
		task.Secrets["ACTIONS_RUNTIME_TOKEN"] = giteaRuntimeToken
	}

	// Register this job's runtime token with the local cache server so that
	// cache requests from the job container can authenticate. The credential
	// is removed when the task finishes, so a leaked token stops working as
	// soon as the job ends rather than remaining valid for the runner's
	// lifetime.
	revokeCache, resultsURL := r.registerCacheForTask(giteaRuntimeToken, cacheCredential(preset, cacheURL), reporter)
	defer revokeCache()

	eventJSON, err := json.Marshal(preset.Event)
	if err != nil {
		return err
	}

	maxLifetime := 3 * time.Hour
	if deadline, ok := ctx.Deadline(); ok {
		maxLifetime = time.Until(deadline)
	}

	// shallow clones the requested ref at depth 1, otherwise 0 means a full clone
	actionCloneDepth := 1
	if r.cfg.Runner.ActionShallowClone != nil && !*r.cfg.Runner.ActionShallowClone {
		actionCloneDepth = 0
	}

	workdirParent := containerWorkdirRoot(r.cfg.Container.WorkdirParent)
	if r.cfg.Container.BindWorkdir {
		// Append the task ID to isolate concurrent jobs from the same repo.
		workdirParent = filepath.Join(workdirParent, strconv.FormatInt(task.Id, 10))
	}
	workdir := filepath.Join(workdirParent, filepath.FromSlash(preset.Repository))
	if runtime.GOOS == "windows" {
		if abs, err := filepath.Abs(workdir); err == nil {
			workdir = abs
		}
	}
	// Without bind_workdir, the workspace path omits the task id; concurrent host-mode jobs
	// for the same repository would share this directory and can race with per-job cleanup.

	r.setResultsService(envs, resultsURL)

	runnerConfig := &runner.Config{
		// On Linux, Workdir will be like "/<parent_directory>/<owner>/<repo>"
		// On Windows, Workdir will be like "\<parent_directory>\<owner>\<repo>"
		Workdir:           workdir,
		BindWorkdir:       r.cfg.Container.BindWorkdir,
		ActionCacheDir:    filepath.FromSlash(r.cfg.Host.WorkdirParent),
		AllocatePTY:       r.cfg.Runner.AllocatePTY,
		ActionOfflineMode: r.cfg.Cache.OfflineMode,
		ActionCloneDepth:  actionCloneDepth,
		NoActionPatch:     r.cfg.Runner.PatchActions != nil && !*r.cfg.Runner.PatchActions,

		ForcePull:            r.cfg.Container.ForcePull,
		ForceRebuild:         r.cfg.Container.ForceRebuild,
		Env:                  envs,
		ProxyEnv:             proxyEnv,
		Secrets:              task.Secrets,
		ExtraMasks:           append(proxyPasswords(r.envs), preset.Token),
		GitHubInstance:       strings.TrimSuffix(r.client.Address(), "/"),
		NoSkipCheckout:       true,
		DisableActEnv:        r.cfg.Runner.SetActEnv != nil && !*r.cfg.Runner.SetActEnv,
		PresetGitHubContext:  preset,
		WorkflowCallInputs:   workflowCallInputs,
		EventJSON:            string(eventJSON),
		ContainerNamePrefix:  fmt.Sprintf("GITEA-ACTIONS-TASK-%d", task.Id),
		ContainerMaxLifetime: maxLifetime,
		CleanWorkdir:         true,
		ContainerNetworkMode: docker_container.NetworkMode(r.cfg.Container.Network),
		CacheContainer:       cacheContainer,
		ContainerNetworkCreateOptions: container.NewDockerNetworkCreateExecutorInput{
			EnableIPv4: r.cfg.Container.NetworkCreateOptions.EnableIPv4,
			EnableIPv6: r.cfg.Container.NetworkCreateOptions.EnableIPv6,
			// so a network this job leaks, if the runner dies before its teardown, can be
			// told apart from one belonging to another runner on the same daemon
			RunnerUUID: r.uuid,
		},
		ContainerOptions:                  r.cfg.Container.Options,
		ServiceReadyTimeout:               r.cfg.Container.ServiceReadyTimeout,
		NoSweep:                           r.cfg.Container.Sweep != nil && !*r.cfg.Container.Sweep,
		ContainerDaemonSocket:             r.cfg.Container.DockerHost,
		Privileged:                        r.cfg.Container.Privileged,
		DefaultActionInstance:             r.getDefaultActionsURL(task),
		DefaultActionInstanceIsSelfHosted: r.isSelfHostedActionsURL(task),
		PlatformPicker:                    platformPicker,
		JobStartedHook:                    r.cfg.Runner.Hooks.JobStarted,
		JobCompletedHook:                  r.cfg.Runner.Hooks.JobCompleted,
		Vars:                              task.Vars,
		ValidVolumes:                      r.cfg.Container.ValidVolumes,
		SharedToolCache:                   r.cfg.Runner.ToolCacheMode == config.ToolCacheModeShared,
		InsecureSkipTLS:                   r.cfg.Runner.Insecure,
		ClientCertFile:                    r.cfg.Runner.ClientCertFile,
		ClientKeyFile:                     r.cfg.Runner.ClientKeyFile,
		InstanceTransport:                 r.instanceTransport,
		RunnerName:                        r.name,
		KubernetesPicker:                  r.kubernetesOptions,
	}

	rr, err := runner.New(runnerConfig)
	if err != nil {
		return err
	}
	executor := rr.NewPlanExecutor(plan)

	reporter.Logf("workflow prepared")

	// add logger recorders
	ctx = common.WithLoggerHandler(ctx, reporter)

	if !logging.Enabled(logging.LevelDebug) {
		ctx = runner.WithJobLoggerFactory(ctx, NullLogger{})
	}

	execErr := executor(ctx)
	reporter.SetOutputs(job.Outputs)

	if r.cfg.Container.BindWorkdir {
		if err := os.RemoveAll(workdirParent); err != nil {
			logging.Warnf("failed to clean up workspace %s: %v", workdirParent, err)
		}
	}

	reporter.StopHeartbeats()
	r.runPostTaskScript(ctx, reporter, task, workdir)

	return execErr
}

func (r *Runner) builtInCacheURL() string {
	if r.cacheHandler == nil {
		return ""
	}
	return r.envs["ACTIONS_CACHE_URL"]
}

func (r *Runner) setResultsService(envs map[string]string, resultsURL string) {
	v2 := resultsURL != "" && (r.cfg.Cache.V2 == nil || *r.cfg.Cache.V2) // all cache.v2 turns off, the bundle edit reaching it is what the artifact actions need too
	if v2 {
		envs[runner.CacheServiceV2Env] = "true"
	} else {
		delete(envs, runner.CacheServiceV2Env)
	}
	// Cache v2 clients read no other variable for it, and some cannot use the instance address.
	if v2 || (resultsURL != "" && r.instanceOutOfReach(envs["ACTIONS_RESULTS_URL"])) {
		envs["ACTIONS_RESULTS_URL"] = resultsURL
	}
}

// These clients keep only the origin they are handed, and reach it on the job's own trust.
func (r *Runner) instanceOutOfReach(instance string) bool {
	parsed, err := url.Parse(instance)
	return err != nil || strings.Trim(parsed.Path, "/") != "" ||
		(r.cfg.Runner.Insecure && parsed.Scheme == "https")
}

// registerCacheForTask tells the cache server to accept requests authenticated
// with the given runtime token for the duration of this task. Returns a
// function the caller must invoke (typically via defer) to revoke the
// credential when the task finishes.
//
// Safe with an empty token (older Gitea did not issue one).
// It also returns the URL to advertise as ACTIONS_RESULTS_URL.
func (r *Runner) registerCacheForTask(token string, cred artifactcache.JobCredential, reporter *report.Reporter) (func(), string) {
	if token == "" || r.cacheHandler == nil {
		return func() {}, ""
	}
	cred.Results = r.envs["ACTIONS_RESULTS_URL"] // the instance as the job would reach it
	cred.InsecureTLS = r.cfg.Runner.Insecure
	revoke := r.cacheHandler.RegisterJob(token, cred)
	if r.cfg.Cache.ExternalServer != "" {
		revokeLocal, revokeExternal := revoke, r.registerExternalCacheJob(token, cred, reporter)
		revoke = func() { revokeLocal(); revokeExternal() }
	}
	return revoke, r.cacheHandler.ResultsURL(cred)
}

// cacheCredential scopes the job's cache entries by ref, as GitHub does.
func cacheCredential(preset *model.GithubContext, publicURL string) artifactcache.JobCredential {
	ref := preset.Ref
	if strings.HasPrefix(preset.EventName, "pull_request") && preset.EventName != "pull_request_target" {
		number, _ := preset.Event["number"].(float64)
		ref = fmt.Sprintf("refs/pull/%d/head", int64(number)) // review events take github.ref from the reviewer's commit_id
	}
	return artifactcache.JobCredential{
		Repo:       preset.Repository,
		Ref:        ref,
		BaseRef:    branchRef(preset.BaseRef),
		DefaultRef: branchRef(model.AsString(model.NestedMapLookup(preset.Event, "repository", "default_branch"))),
		PublicURL:  publicURL,
	}
}

func branchRef(branch string) string {
	if branch == "" {
		return ""
	}
	return "refs/heads/" + branch
}

// registerExternalCacheJob POSTs to the remote cache-server's control-plane.
// Failures are logged but not fatal: if registration fails, the cache will
// 401 the job's requests — better than failing the whole task for a cache
// outage. The warning is mirrored to the job log so users can see why their
// cache calls 401, instead of having to read the runner daemon's stderr.
func (r *Runner) registerExternalCacheJob(token string, cred artifactcache.JobCredential, reporter *report.Reporter) func() {
	base := strings.TrimRight(r.cfg.Cache.ExternalServer, "/")
	if err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, struct {
		Token string `json:"token"`
		artifactcache.JobCredential
	}{token, cred}); err != nil {
		logging.Warnf("cache external_server register failed (%s): %v", base, err)
		if reporter != nil {
			reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf(
				"cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err)))
		}
	}
	return func() {
		if err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret,
			map[string]any{"token": token}); err != nil {
			logging.Warnf("cache external_server revoke failed (%s): %v", base, err)
			if reporter != nil {
				reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf(
					"cache external_server revoke failed (%s): %v", base, err)))
			}
		}
	}
}

func postInternalCache(url, secret string, body any) error {
	buf, err := json.Marshal(body)
	if err != nil {
		return err
	}
	req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(buf))
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+secret)
	req.Header.Set("Content-Type", "application/json")
	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		return fmt.Errorf("status %d", resp.StatusCode)
	}
	return nil
}

func (r *Runner) RunningCount() int64 {
	return r.runningCount.Load()
}

// CanAcceptTask checks local admission conditions without consuming a task. It is
// called only from the poll loop, so the cached health fields need no lock.
func (r *Runner) CanAcceptTask(ctx context.Context) (bool, string) {
	if !r.cfg.HealthCheck.Enabled {
		return true, "ok"
	}

	if r.RunningCount() > 0 {
		if !r.healthStatusSet {
			return true, "health checks deferred while jobs are running"
		}
		return r.healthStatusReady, r.healthStatusReason
	}

	if ready, reason := checkFreeDisk(r.cfg); !ready {
		r.setHealthStatus(ready, reason)
		return false, reason
	}
	ready, reason := r.checkConfiguredHealth(ctx)
	r.setHealthStatus(ready, reason)
	return ready, reason
}

func (r *Runner) setHealthStatus(ready bool, reason string) {
	r.healthStatusSet = true
	r.healthStatusReady = ready
	r.healthStatusReason = reason
}

// checkFreeDisk evaluates the configured task-admission disk threshold.
func checkFreeDisk(cfg *config.Config) (bool, string) {
	root := cfg.Host.WorkdirParent
	if cfg.Container.BindWorkdir {
		root = containerWorkdirRoot(cfg.Container.WorkdirParent)
	}
	root = nearestExistingPath(root)
	available, err := disk.FreeBytes(root)
	if err != nil {
		return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
	}
	availableMB := available / (1024 * 1024)
	if availableMB < uint64(cfg.HealthCheck.MinFreeDiskSpaceMB) {
		return false, fmt.Sprintf("low disk space on %s: %d MiB available, %d MiB required", root, availableMB, cfg.HealthCheck.MinFreeDiskSpaceMB)
	}
	return true, "ok"
}

func containerWorkdirRoot(parent string) string {
	if filepath.IsAbs(parent) {
		return filepath.Clean(parent)
	}
	return filepath.Join(string(filepath.Separator), parent)
}

func nearestExistingPath(path string) string {
	for path != "" {
		if _, err := os.Stat(path); err == nil {
			return path
		}
		parent := filepath.Dir(path)
		if parent == path {
			return parent
		}
		path = parent
	}
	return "."
}

func (r *Runner) Declare(ctx context.Context, labels []string) (*runnerv1.DeclareResponse, error) {
	ctx, info := connect.NewClientContext(ctx)
	resp, err := r.client.Declare(ctx, &runnerv1.DeclareRequest{
		Version:      ver.Version(),
		Labels:       labels,
		Capabilities: RunnerCapabilities(),
	})
	if err == nil {
		r.capabilities = strings.TrimSpace(info.ResponseHeader().Get("X-Gitea-Actions-Capabilities"))
	}
	return resp, err
}

// minFreeDisk keeps the cache from growing past the point where the runner stops taking
// work, but only when health checks are on, since the key is documented as opt-in.
func minFreeDisk(cfg *config.Config) int64 {
	if !cfg.HealthCheck.Enabled {
		return 0
	}
	return cfg.HealthCheck.MinFreeDiskSpaceMB * 1024 * 1024
}

// CachePolicy maps the cache config onto the cache server's own type, in bytes not MiB.
func CachePolicy(cfg *config.Config) artifactcache.Policy {
	return artifactcache.Policy{
		Retention:     cfg.Cache.Retention,
		RepoSizeLimit: int64(cfg.Cache.RepoSizeLimit),
		SizeLimit:     int64(cfg.Cache.SizeLimit),
		SweepInterval: cfg.Cache.SweepInterval,
		MinFreeDisk:   minFreeDisk(cfg),
	}
}

func cacheS3(cfg *config.Config) *artifactcache.S3Options {
	s3 := cfg.Cache.S3
	if s3 == nil {
		return nil
	}
	return &artifactcache.S3Options{
		Endpoint: s3.Endpoint, Region: s3.Region, Bucket: s3.Bucket, Prefix: s3.Prefix,
		PathStyle: s3.PathStyle, AccessKey: s3.AccessKey, SecretKey: s3.SecretKey,
	}
}

// warnIgnoredCachePolicy flags eviction settings configured on a runner that does not apply them.
func warnIgnoredCachePolicy(cfg *config.Config) {
	defaults := config.DefaultCache()
	limits := cfg.Cache.Retention != defaults.Retention || cfg.Cache.RepoSizeLimit != defaults.RepoSizeLimit ||
		cfg.Cache.SizeLimit != defaults.SizeLimit
	switch {
	case cfg.Cache.ExternalServer != "" && (limits || cfg.Cache.SweepInterval != defaults.SweepInterval):
		logging.Warn("cache eviction settings are ignored when cache.external_server is set; configure them on that server instead")
	case cfg.Cache.S3 != nil && limits:
		logging.Warn("cache eviction settings are ignored when cache.s3 is set; use lifecycle rules on the bucket instead")
	}
}

// warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server.
func warnIgnoredCacheSecret(cfg *config.Config) {
	if cfg.Cache.ExternalServer != "" || cfg.Cache.ExternalSecret == "" {
		return
	}
	// LoadDefault resolves external_secret_file into ExternalSecret, so report whichever key the operator actually wrote.
	key := "cache.external_secret"
	if cfg.Cache.ExternalSecretFile != "" {
		key = "cache.external_secret_file"
	}
	logging.Warnf("%s is set but cache.external_server is not; the built-in cache server does not use a shared secret, so the value is ignored", key)
}

// cacheForJob returns the cache address for the job, and the container to attach to its network to reach it there.
func (r *Runner) cacheForJob(job *model.Job, platform string) (string, string) {
	jobContainer := job.Container()
	cacheContainer := ""
	if !strings.HasPrefix(platform, labels.SchemeKubernetes+"://") && (jobContainer != nil && jobContainer.Image != "" || platform != labels.SelfHostedPlatform) {
		cacheContainer = r.isolatedCacheContainer()
	}
	if cacheContainer == "" {
		return strings.TrimSuffix(r.builtInCacheURL(), "/"), ""
	}
	cacheURL, _ := url.Parse(r.cacheHandler.ExternalURL())
	cacheURL.Host = net.JoinHostPort(cacheContainer, cacheURL.Port())
	return cacheURL.String(), cacheContainer
}

func (r *Runner) detectIsolatedCacheContainer() string {
	if r.cacheHandler == nil {
		return ""
	}
	addr, err := netip.ParseAddr(hostOf(r.cacheHandler.ExternalURL()))
	if err != nil {
		return ""
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	cacheContainer, err := container.IsolatedCacheContainer(ctx, addr, r.cfg.Container.Network)
	if err != nil {
		logging.Warnf("cannot check whether jobs reach the cache server: %v", err)
	}
	return cacheContainer
}

// taskLogFields returns key/value pairs for logging.With.
func taskLogFields(task *runnerv1.Task, job string) []any {
	taskContext := task.GetContext().GetFields()
	fields := []any{
		"task_id", task.Id,
		"job", job,
		"repository", taskContext["repository"].GetStringValue(),
		"run_id", taskContext["run_id"].GetStringValue(),
		"run_number", taskContext["run_number"].GetStringValue(),
	}
	if parts := strings.Split(taskContext["gitea_runtime_token"].GetStringValue(), "."); len(parts) == 3 {
		var claims struct{ JobID int64 } // Gitea's actionsClaims fields are untagged, so the claim key is "JobID"
		payload, err := base64.RawURLEncoding.DecodeString(parts[1])
		if err == nil && json.Unmarshal(payload, &claims) == nil && claims.JobID > 0 {
			fields = append(fields, "job_id", claims.JobID)
		}
	}
	return fields
}