fix(runner): correct cgroup controller expression

.gitea/workflows/pod-smoke.yml

@@ -1,5 +1,7 @@
+---
 name: dynamic Pod smoke test
 
+# yamllint disable-line rule:truthy
 on:
   workflow_dispatch:
 
@@ -17,3 +19,24 @@ jobs:
             -socketPath /run/spire/agent-sockets/spire-agent.sock \
             >/dev/null
           test "$(id -u)" = 2000
+
+      - name: Build and run image
+        shell: bash
+        run: |
+          set -euo pipefail
+          findmnt /var/lib/docker
+          context=$(mktemp -d)
+          cleanup() {
+            docker image rm --force pod-docker-smoke:test \
+              >/dev/null 2>&1 || true
+            rm -rf -- "$context"
+          }
+          trap cleanup EXIT
+          printf '%s\n' \
+            'FROM alpine:3.22' \
+            'RUN printf pod-docker-ok >/result' \
+            >"$context/Dockerfile"
+          docker build --tag pod-docker-smoke:test "$context"
+          output=$(docker run --rm pod-docker-smoke:test cat /result)
+          test "$output" = pod-docker-ok
+          test "$(docker info --format '{{.Driver}}')" = overlay2

.gitea/workflows/publish-images.yml

@@ -34,17 +34,27 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
       - name: Test source
         shell: bash
         run: |
           set -euo pipefail
           go test ./...
           go vet ./...
-          python3 -m pip install --break-system-packages -e '.[test]'
+          python3 -m venv /tmp/test-venv
+          source /tmp/test-venv/bin/activate
+          python3 -m pip install -e '.[test]'
           pytest -q
           python3 -m compileall -q src tests
-          apt-get update
-          apt-get install --yes --no-install-recommends shellcheck
+          sudo apt-get update
+          sudo apt-get install --yes --no-install-recommends shellcheck
           shellcheck scripts/*
 
       - name: Build and publish
@@ -53,6 +63,10 @@ jobs:
           set -euo pipefail
           set +x
 
+          # Bootstrap the image that first introduces automatic Docker setup.
+          # Once deployed, the job-started hook makes this an idempotent no-op.
+          sudo scripts/setup-job-docker
+
           : "${GITHUB_SHA:?GITHUB_SHA is required}"
           image_tag="sha-${GITHUB_SHA}"
           docker_config=$(mktemp -d)

.gitea/workflows/test.yml

@@ -17,7 +17,7 @@ jobs:
       - run: go vet ./...
 
   python:
-    runs-on: self-hosted
+    runs-on: [self-hosted, pod]
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -28,7 +28,7 @@ jobs:
       - run: python -m compileall -q src
 
   shell:
-    runs-on: self-hosted
+    runs-on: [self-hosted, pod]
     steps:
       - uses: actions/checkout@v4
       - run: |

.gitea/workflows/vm-kind-smoke.yml

@@ -1,5 +1,7 @@
+---
 name: VM kind smoke
 
+# yamllint disable-line rule:truthy
 on:
   workflow_dispatch:
 
@@ -7,25 +9,123 @@ jobs:
   kind:
     runs-on: [self-hosted, vm]
     steps:
+      - name: Prepare nested kubelet device
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [[ ! -e /dev/kmsg ]]; then
+            sudo mknod /dev/kmsg c 1 11
+          fi
+
       - name: Verify Docker
-        run: docker info
+        shell: bash
+        run: |
+          set -euo pipefail
+          findmnt /var/lib/docker
+          docker info
+          echo '### runner cgroup'
+          cat /proc/self/cgroup
+          cat /sys/fs/cgroup/cgroup.type
+          cat /sys/fs/cgroup/cgroup.controllers
+          cat /sys/fs/cgroup/cgroup.subtree_control
+          echo '### nested private cgroup namespace'
+          docker run --rm --privileged --cgroupns=private alpine:3.22 \
+            sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
+          echo '### nested host cgroup namespace'
+          docker run --rm --privileged --cgroupns=host alpine:3.22 \
+            sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
 
       - name: Install kind
         shell: bash
         run: |
           set -euo pipefail
           version=v0.33.0
+          base_url="https://kind.sigs.k8s.io/dl/${version}"
           curl --fail --location --silent --show-error \
-            --output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
+            --output /tmp/kind "${base_url}/kind-linux-amd64"
           curl --fail --location --silent --show-error \
-            --output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
-          printf '%s  %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
+            --output /tmp/kind.sha256sum \
+            "${base_url}/kind-linux-amd64.sha256sum"
+          checksum=$(cut -d ' ' -f1 /tmp/kind.sha256sum)
+          printf '%s  %s\n' "$checksum" /tmp/kind | sha256sum --check
           chmod 0755 /tmp/kind
 
       - name: Create and delete kind cluster
         shell: bash
         run: |
           set -euo pipefail
-          trap '/tmp/kind delete cluster --name smoke' EXIT
-          /tmp/kind create cluster --name smoke --wait 180s
+          diagnose_and_cleanup() {
+            status=$?
+            node=smoke-control-plane
+            if (( status != 0 )) && docker inspect "$node" >/dev/null 2>&1; then
+              echo '::group::kind node inspect'
+              docker inspect "$node"
+              echo '::endgroup::'
+              echo '::group::kind node logs'
+              docker logs "$node" 2>&1 || true
+              echo '::endgroup::'
+              echo '::group::kind node guest state'
+              docker exec "$node" bash -c '
+                set +e
+                echo "### pid 1"
+                ps -p 1 -o pid,ppid,user,stat,comm,args
+                cat /proc/1/status
+                echo "### cgroup"
+                cat /proc/1/cgroup
+                findmnt -R /sys/fs/cgroup
+                stat -fc "%T %a" /sys/fs/cgroup
+                echo "### systemd"
+                systemctl --no-pager --failed
+                systemctl --no-pager status \
+                  multi-user.target containerd.service kubelet.service
+                echo "### CRI containers"
+                endpoint=unix:///run/containerd/containerd.sock
+                crictl --runtime-endpoint "$endpoint" ps --all
+                for id in $(
+                  crictl --runtime-endpoint "$endpoint" ps --all --quiet
+                ); do
+                  echo "### CRI container $id"
+                  crictl --runtime-endpoint "$endpoint" inspect "$id"
+                  crictl --runtime-endpoint "$endpoint" logs "$id"
+                done
+                echo "### containerd metadata"
+                timeout 10 ctr --namespace k8s.io containers list
+                timeout 10 ctr --namespace k8s.io snapshots list
+                echo "### runtime process stacks"
+                ps -e -o pid,ppid,stat,wchan:32,comm,args
+                for pid in $(pidof containerd containerd-shim-runc-v2); do
+                  echo "### kernel stack $pid"
+                  cat "/proc/$pid/stack"
+                done
+                kill -USR1 "$(pidof containerd)"
+                sleep 2
+                journalctl --no-pager -b -n 500
+              ' 2>&1 || true
+              echo '::endgroup::'
+            fi
+            /tmp/kind delete cluster --name smoke || true
+            exit "$status"
+          }
+          trap diagnose_and_cleanup EXIT
+          # Nested Kata + Docker + kind cold starts can take longer than
+          # kubeadm's one-minute API-call default even after the static pods
+          # have been accepted. Give the API server enough time to become
+          # responsive before kubeadm creates its initial RBAC objects.
+          cat >/tmp/kind-config.yml <<'EOF'
+          kind: Cluster
+          apiVersion: kind.x-k8s.io/v1alpha4
+          nodes:
+            - role: control-plane
+              kubeadmConfigPatches:
+                - |
+                  apiVersion: kubeadm.k8s.io/v1beta4
+                  kind: InitConfiguration
+                  timeouts:
+                    kubernetesAPICall: 5m0s
+          EOF
+          /tmp/kind create cluster \
+            --name smoke \
+            --config /tmp/kind-config.yml \
+            --wait 300s \
+            --retain
           /tmp/kind get clusters | grep -Fx smoke

.gitea/workflows/vm-runtime-smoke.yml

@@ -0,0 +1,36 @@
+---
+name: VM runtime smoke
+
+# yamllint disable-line rule:truthy
+on:
+  workflow_dispatch:
+
+jobs:
+  runtime:
+    runs-on: [self-hosted, vm]
+    steps:
+      - name: Verify workload identity socket
+        shell: bash
+        run: |
+          set -euo pipefail
+          : "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
+          socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
+          test -S "$socket_path"
+
+      - name: Verify Docker daemon
+        shell: bash
+        run: |
+          set -euo pipefail
+          format='{{json .ServerVersion}} {{json .Driver}}'
+          format="$format {{json .CgroupVersion}}"
+          docker info --format "$format"
+
+      - name: Run and clean nested container
+        shell: bash
+        run: |
+          set -euo pipefail
+          name=vm-runtime-smoke
+          trap 'docker rm --force "$name" >/dev/null 2>&1 || true' EXIT
+          output=$(docker run --name "$name" alpine:3.22 /bin/sh -c \
+            'test "$(uname -m)" = x86_64 && printf vm-runtime-ok')
+          test "$output" = vm-runtime-ok

README.md

@@ -15,10 +15,15 @@ runs-on: [self-hosted, vm]
 只执行一个 job，并在 job 结束后连同本地状态一起销毁。完整的设计约束见
 [`docs/design-principles.md`](docs/design-principles.md)。
 
+集成期间可将 `VM_RUNNER_LABEL=vm-dev`，只接取显式使用
+`runs-on: [self-hosted, vm-dev]` 的测试任务；生产 `vm` job 将保持在 Gitea pending，
+不会在 backend 修复过程中继续涌入。
+
 目标 Go controller 组件：
 
 - `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment
-  持久化到 JetStream；同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
+  持久化到 JetStream；一个 registration 下按配置启动多个并发 `FetchTask` goroutine，
+  同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
 - `pod-worker`：直接在 homelab Kubernetes 创建一次性 Pod。
 - `vm-worker`：通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。
 - 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker，因为 facade
@@ -31,14 +36,16 @@ runs-on: [self-hosted, vm]
   entry；不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
   [`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
 - Pod executor：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用
-  host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
-  支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。
+  host executor。Runner 固定在支持原生 job hooks 的 3.x 版本，在 workflow 第一步前
+  等待实际任务对应的 SVID，并启动 job-local Docker daemon；workflow 可直接使用与
+  GitHub-hosted runner 相同的 Docker/BuildKit action。
 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个
   动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
 
 Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于
-VM/Kata workload；两个 backend 使用独立 durable consumer，任一执行层故障不会阻塞另一条
-部署。长期 RunnerService 协议路线见
+VM/Kata workload；两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据
+`runs-on` 进入对应池，池满时留在 JetStream pending，不会创建超出容量的 workload；任一
+执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见
 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
 
 ## 开发

cmd/gitea-dynamic-runner/controller.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"net/http"
 	"os"
 	"slices"
@@ -15,8 +15,14 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"golang.org/x/sync/errgroup"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/backendpool"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/giteaactions"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/opensandboxbackend"
@@ -43,7 +49,7 @@ type controllerConfig struct {
 	PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
 	SPIREAgentID                                                        string
 	PodExecutorUID, PodCapacity                                         int
-	OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool                  string
+	OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool, VMRunnerLabel   string
 	VMTimeout, VMCapacity                                               int
 }
 
@@ -86,7 +92,8 @@ func runController(ctx context.Context) error {
 		return err
 	}
 	registry := runnerfacade.NewRegistry()
-	gate := taskscheduler.NewSingleFlightGate()
+	podPool := backendpool.New(config.PodCapacity)
+	vmPool := backendpool.New(config.VMCapacity)
 	var podExecutorBackend *podbackend.Backend
 	var vmExecutorBackend *opensandboxbackend.Backend
 	giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
@@ -101,6 +108,7 @@ func runController(ctx context.Context) error {
 				if err := podExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
 					return err
 				}
+				podPool.Release(assignment.ID)
 			case taskassignment.BackendVM:
 				if vmExecutorBackend == nil {
 					return errors.New("VM lifecycle backend is not configured")
@@ -108,8 +116,8 @@ func runController(ctx context.Context) error {
 				if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
 					return err
 				}
+				vmPool.Release(assignment.ID)
 			}
-			gate.Release()
 			return nil
 		},
 	}
@@ -123,15 +131,27 @@ func runController(ctx context.Context) error {
 		labels = append(labels, string(taskassignment.BackendPod))
 	}
 	if slices.Contains(config.Components, controller.VMWorker) {
-		labels = append(labels, string(taskassignment.BackendVM))
+		labels = append(labels, config.VMRunnerLabel)
 	}
 	poller := taskscheduler.Poller{
-		Client: giteaClient, Gate: gate,
+		Client: giteaClient,
 		Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{
 			JetStream: producerJS, SubjectBase: config.SubjectBase,
 		}},
-		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels},
-		OnError: func(err error) { log.Printf("scheduler: %v", err) },
+		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
+		OnError: func(err error) { slog.Error("scheduler error", "component", "scheduler", "error", err) },
+	}
+	kubernetesConfig, err := rest.InClusterConfig()
+	if err != nil {
+		return fmt.Errorf("load leader election Kubernetes config: %w", err)
+	}
+	kubernetesClient, err := kubernetes.NewForConfig(kubernetesConfig)
+	if err != nil {
+		return fmt.Errorf("create leader election Kubernetes client: %w", err)
+	}
+	leaderIdentity := strings.TrimSpace(os.Getenv("HOSTNAME"))
+	if leaderIdentity == "" {
+		return errors.New("HOSTNAME is required for scheduler leader election")
 	}
 	facadeServer := runnerfacade.Server{
 		Facade: facade, ListenAddress: config.FacadeListen, TrustDomain: config.TrustDomain,
@@ -141,7 +161,9 @@ func runController(ctx context.Context) error {
 		controller.Scheduler: runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return facadeServer.Run(groupContext) })
-			group.Go(func() error { return poller.Run(groupContext) })
+			group.Go(func() error {
+				return runSchedulerLeader(groupContext, kubernetesClient, config.PodNamespace, leaderIdentity, poller.Run)
+			})
 			return group.Wait()
 		}),
 	}
@@ -158,11 +180,23 @@ func runController(ctx context.Context) error {
 			SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
 		}}
 		podExecutorBackend = &backend
-		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
+		assignments, err := backend.RecoverAssignments(ctx)
 		if err != nil {
 			return err
 		}
-		lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
+		for _, assignment := range assignments {
+			podPool.Restore(assignment.ID)
+			if err := registry.RecoverClaimed(assignment); err != nil {
+				return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
+			}
+		}
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendPod)}, registry, podPool)
+		if err != nil {
+			return err
+		}
+		lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) {
+			slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendPod, "error", err)
+		}}
 		components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return component.Run(groupContext) })
@@ -178,11 +212,23 @@ func runController(ctx context.Context) error {
 			Env:        map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
 		}}
 		vmExecutorBackend = &backend
-		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
+		assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
 		if err != nil {
 			return err
 		}
-		lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
+		for _, assignment := range assignments {
+			vmPool.Restore(assignment.ID)
+			if err := registry.RecoverClaimed(assignment); err != nil {
+				return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
+			}
+		}
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendVM)}, registry, vmPool)
+		if err != nil {
+			return err
+		}
+		lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) {
+			slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendVM, "error", err)
+		}}
 		components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return component.Run(groupContext) })
@@ -193,6 +239,50 @@ func runController(ctx context.Context) error {
 	return controller.Run(ctx, config.Components, components)
 }
 
+func runSchedulerLeader(ctx context.Context, client kubernetes.Interface, namespace, identity string, run func(context.Context) error) error {
+	if client == nil || namespace == "" || identity == "" || run == nil {
+		return errors.New("leader election client, namespace, identity, and scheduler are required")
+	}
+	electionContext, cancel := context.WithCancel(ctx)
+	defer cancel()
+	result := make(chan error, 1)
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{Name: "dynamic-runner-scheduler", Namespace: namespace},
+		Client:    client.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+	elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock: lock, LeaseDuration: 15 * time.Second, RenewDeadline: 10 * time.Second, RetryPeriod: 2 * time.Second,
+		ReleaseOnCancel: true,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(leaderContext context.Context) {
+				result <- run(leaderContext)
+				cancel()
+			},
+			OnStoppedLeading: func() {
+				if ctx.Err() == nil {
+					select {
+					case result <- errors.New("scheduler leadership lost"):
+					default:
+					}
+				}
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("configure scheduler leader election: %w", err)
+	}
+	go elector.Run(electionContext)
+	select {
+	case err := <-result:
+		return err
+	case <-ctx.Done():
+		return nil
+	}
+}
+
 func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, error) {
 	options := []nats.Option{nats.Name(clientName), nats.UserInfo(user, password)}
 	if caFile != "" {
@@ -201,18 +291,28 @@ func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn,
 	return nats.Connect(server, options...)
 }
 
-func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims) (controller.Component, error) {
+func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) {
 	consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity)
 	if err != nil {
 		return nil, err
 	}
 	return assignmentqueue.ConsumerComponent{
 		Consumer: consumer, Capacity: capacity,
-		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims},
-		OnError:   func(err error) { log.Printf("%s worker: %v", backend, err) },
+		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission, OnEvent: func(event assignmentqueue.Event) {
+			slog.Info("assignment transition", "component", "worker", "event", event.Name, "backend", event.Backend, "assignment", event.AssignmentID, "retry_delay", event.RetryDelay)
+		}},
+		OnError: func(err error) {
+			slog.Error("assignment processing error", "component", "worker", "backend", backend, "error", err)
+		},
 	}, nil
 }
 
+func workerEventLogger(backend taskassignment.Backend) func(taskworker.Event) {
+	return func(event taskworker.Event) {
+		slog.Info("executor transition", "component", "worker", "event", event.Name, "backend", backend, "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase)
+	}
+}
+
 func loadControllerConfig() (controllerConfig, error) {
 	selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
 	if err != nil {
@@ -259,7 +359,7 @@ func loadControllerConfig() (controllerConfig, error) {
 		FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
 		PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),
 		SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), SPIREAgentID: os.Getenv("SPIRE_AGENT_ID"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4),
-		OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
+		OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMRunnerLabel: env("VM_RUNNER_LABEL", "vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
 	}
 	if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
 		return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required")
@@ -271,6 +371,9 @@ func loadControllerConfig() (controllerConfig, error) {
 		return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
 	}
 	if slices.Contains(selection, controller.VMWorker) {
+		if config.VMRunnerLabel != "vm" && config.VMRunnerLabel != "vm-dev" {
+			return controllerConfig{}, errors.New("VM_RUNNER_LABEL must be vm or vm-dev")
+		}
 		if config.OpenSandboxURL == "" {
 			return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
 		}

cmd/gitea-dynamic-runner/controller_test.go

@@ -56,6 +56,7 @@ func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) {
 	t.Setenv("RUNNER_FACADE_URL", "https://facade:8443")
 	t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/controller")
 	t.Setenv("OPENSANDBOX_API", "http://opensandbox.internal")
+	t.Setenv("VM_RUNNER_LABEL", "vm-dev")
 	if _, err := loadControllerConfig(); err == nil {
 		t.Fatal("expected missing OpenSandbox API key file error")
 	}

cmd/gitea-dynamic-runner/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"os"
 	"os/signal"
 	"syscall"
@@ -12,8 +13,9 @@ import (
 )
 
 func main() {
+	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
 	if err := run(); err != nil {
-		fmt.Fprintln(os.Stderr, err)
+		slog.Error("runner stopped", "error", err)
 		os.Exit(1)
 	}
 }

container/runner.Dockerfile

@@ -29,6 +29,7 @@ COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
 COPY config/runner.yaml /etc/gitea-runner/config.yaml
 COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
 COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner
+COPY --chmod=0755 scripts/setup-job-docker /usr/local/libexec/setup-job-docker
 
 VOLUME ["/data"]
 ENV HOME=/home/runner

docs/opensandbox-runner.md

@@ -40,9 +40,11 @@ UID attestation 的临时 Agent 失去父级。
 
 `ci-vm` 使用 `kata-clh-runtime-rs`；`ci-pod` 使用默认 runc。两者都要求：
 
-- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate；
-- runner UID 2000，SPIRE Agent 与 privileged dockerd 使用不同 UID；
-- Docker socket 通过 group 2000 共享，Docker 数据仅存在于 sandbox emptyDir；
+- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI、Docker 工具和
+  identity gate；Pod 与 VM backend 使用同一个镜像；
+- runner UID 2000；SPIRE Agent 独立运行；job-started hook 在第一步 workflow 之前
+  启动 job-local Docker daemon，业务 workflow 不负责 runner 基础设施初始化；
+- Kata VM 中 Docker 数据使用 guest 内的 loop-backed ext4，并随 sandbox 一起删除；
 - `self-hosted` 必须是所有 runner labels 的前缀；
 - ephemeral/once runner 完成一项任务后退出。
 

docs/runner-protocol-roadmap.md

@@ -3,8 +3,9 @@
 ## 目标
 
 长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
-协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；它只在后端存在可用容量
-时领取 task，然后将该 task 交给一个一次性 Pod 或 microVM 执行。
+协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；单个 registration 内按总
+配置容量启动多个 `FetchTask` goroutine，再将 task 按 `runs-on` 交给 Pod 或 VM 的独立
+容量池，由一次性 Pod 或 microVM 执行。
 
 ```text
 Gitea RunnerService
@@ -47,7 +48,10 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
 ## 设计约束
 
 - 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
-- scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。
+- scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`，不为并发槽位重复注册；
+  `POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。
+- task 领取并持久化后按 backend 进入独立 durable consumer；对应容量池已满时延迟 NAK，
+  assignment 保持 JetStream pending，且不得创建超出配置容量的 workload。
 - scheduler Declare 后使用 RunnerService 长轮询；一旦 FetchTask 返回已分配 task，在
   JetStream publish 成功前只重试该 assignment，不领取下一项。
 - 每个 executor 只执行一个 task，完成后销毁。
@@ -72,17 +76,19 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
 - executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后，facade 在后端
   metadata 写入持久 terminal marker；backend reconciler 仅在执行环境也进入终态后清理，
   从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
-- pod 与 vm 使用独立 durable consumer 和并发上限。consumer 只负责将 assignment
+- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment
   幂等落到后端；executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得
   Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。
+- admission pool 只保存可重建的并发状态：启动时从 Pod labels/annotations 或 OpenSandbox
+  metadata 恢复非终态 assignment，terminal update 持久化成功后释放槽位，不引入新存储。
 - assignment ACK 后的运行、结果回报和清理由 backend reconciler 根据 Kubernetes、
   OpenSandbox 与 Gitea 的事实状态驱动，不继续占用 JetStream delivery。
 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整
   task 写入 Pod annotation、OpenSandbox metadata 或环境变量。
 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。
-- 首次生产 canary 使用 controller 进程内 single-flight gate：只有官方 runner 的终态
-  `UpdateTask` 已被 Gitea 接受后才允许 Fetch 下一条任务。它把未知故障收敛为停止领取，
-  而不是在 backlog 下连续创建 executor；后续容量调度必须以 backend 实际运行资源为准。
+- scheduler 在 assignment 持久化到 JetStream 后即可继续领取；Pod 与 VM 分别由 durable
+  consumer 的 capacity 限制并发，不共享全局执行槽位。未知后端故障由对应 consumer 的
+  NAK/redelivery 收敛，不能阻塞另一种 backend。
 - 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生
   创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。
 

go.mod

@@ -45,6 +45,7 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/sirupsen/logrus v1.10.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect

go.sum

@@ -112,6 +112,8 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W
 go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
 go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
 go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=

internal/assignmentqueue/jetstream.go

@@ -57,6 +57,11 @@ type Claims interface {
 	WaitClaimed(context.Context, string) error
 }
 
+type Admission interface {
+	Acquire(string) bool
+	Release(string)
+}
+
 // Message is the subset of jetstream.Msg needed by one reconciliation.
 type Message interface {
 	Data() []byte
@@ -70,30 +75,60 @@ type Processor struct {
 	TrustDomain  string
 	Accepter     Accepter
 	Claims       Claims
+	Admission    Admission
 	RetryDelay   time.Duration
 	ClaimTimeout time.Duration
+	OnEvent      func(Event)
+}
+
+// Event describes a non-sensitive assignment handoff transition. It never
+// contains task payloads, credentials, capabilities, or workload identities.
+type Event struct {
+	Name         string
+	AssignmentID string
+	Backend      taskassignment.Backend
+	RetryDelay   time.Duration
+}
+
+func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) {
+	if p.OnEvent != nil {
+		p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay})
+	}
 }
 
 func (p Processor) Process(ctx context.Context, message Message) error {
-	if p.Accepter == nil || p.Claims == nil {
-		return errors.New("assignment accepter and claim registry are required")
+	if p.Accepter == nil || p.Claims == nil || p.Admission == nil {
+		return errors.New("assignment accepter, claim registry, and backend admission pool are required")
 	}
 	assignment, err := taskassignment.Unmarshal(message.Data(), p.TrustDomain)
 	if err != nil {
 		return errors.Join(err, message.TermWithReason("invalid assignment"))
 	}
+	p.event("received", assignment, 0)
 	if _, err := p.Claims.Offer(assignment); err != nil {
 		return errors.Join(err, message.TermWithReason("conflicting assignment"))
 	}
+	if !p.Admission.Acquire(assignment.ID) {
+		delay := p.RetryDelay
+		if delay <= 0 {
+			delay = 2 * time.Second
+		}
+		p.event("capacity_wait", assignment, delay)
+		return message.NakWithDelay(delay)
+	}
+	p.event("capacity_acquired", assignment, 0)
 	accepted, err := p.Accepter.Accept(ctx, assignment)
 	if err != nil {
+		p.Admission.Release(assignment.ID)
 		delay := p.RetryDelay
 		if delay <= 0 {
 			delay = 15 * time.Second
 		}
+		p.event("backend_retry", assignment, delay)
 		return errors.Join(err, message.NakWithDelay(delay))
 	}
 	if accepted {
+		p.event("backend_ready", assignment, 0)
 		timeout := p.ClaimTimeout
 		if timeout <= 0 {
 			timeout = 4 * time.Minute
@@ -106,17 +141,21 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 			if delay <= 0 {
 				delay = 2 * time.Second
 			}
+			p.event("claim_timeout", assignment, delay)
 			return errors.Join(err, message.NakWithDelay(delay))
 		}
+		p.event("runner_claimed", assignment, 0)
 		if err := message.DoubleAck(ctx); err != nil {
 			return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
 		}
+		p.event("acked", assignment, 0)
 		return nil
 	}
 	delay := p.RetryDelay
 	if delay <= 0 {
 		delay = 2 * time.Second
 	}
+	p.event("backend_pending", assignment, delay)
 	return message.NakWithDelay(delay)
 }
 
@@ -153,7 +192,7 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB
 		AckPolicy:     jetstream.AckExplicitPolicy,
 		AckWait:       5 * time.Minute,
 		MaxAckPending: capacity,
-		MaxDeliver:    20,
+		MaxDeliver:    1000,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)

internal/assignmentqueue/jetstream_test.go

@@ -61,6 +61,28 @@ type fakeClaims struct {
 	claimed bool
 }
 
+type fakeAdmission struct {
+	allowed  bool
+	active   map[string]bool
+	released int
+}
+
+func (a *fakeAdmission) Acquire(assignmentID string) bool {
+	if !a.allowed {
+		return false
+	}
+	if a.active == nil {
+		a.active = make(map[string]bool)
+	}
+	a.active[assignmentID] = true
+	return true
+}
+
+func (a *fakeAdmission) Release(assignmentID string) {
+	delete(a.active, assignmentID)
+	a.released++
+}
+
 func (c *fakeClaims) Offer(taskassignment.Assignment) (<-chan struct{}, error) {
 	ready := make(chan struct{})
 	if c.claimed {
@@ -104,18 +126,28 @@ func encodedAssignment(t *testing.T) []byte {
 
 func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
 	message := &fakeMessage{data: encodedAssignment(t)}
-	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}}
+	var events []Event
+	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}, OnEvent: func(event Event) { events = append(events, event) }}
 	if err := processor.Process(context.Background(), message); err != nil {
 		t.Fatal(err)
 	}
 	if message.acked != 1 || message.nacked != 0 {
 		t.Fatalf("message = %#v", message)
 	}
+	want := []string{"received", "capacity_acquired", "backend_ready", "runner_claimed", "acked"}
+	if len(events) != len(want) {
+		t.Fatalf("events = %#v", events)
+	}
+	for index := range want {
+		if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod {
+			t.Fatalf("event[%d] = %#v", index, events[index])
+		}
+	}
 }
 
 func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
 	message := &fakeMessage{data: encodedAssignment(t)}
-	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, RetryDelay: 2 * time.Second}
+	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, Admission: &fakeAdmission{allowed: true}, RetryDelay: 2 * time.Second}
 	if err := processor.Process(context.Background(), message); err != nil {
 		t.Fatal(err)
 	}
@@ -126,10 +158,12 @@ func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
 
 func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) {
 	retry := &fakeMessage{data: encodedAssignment(t)}
+	admission := &fakeAdmission{allowed: true}
 	processor := Processor{
 		TrustDomain: "ddupan.top",
 		Accepter:    &fakeAccepter{err: errors.New("backend unavailable")},
 		Claims:      &fakeClaims{},
+		Admission:   admission,
 		RetryDelay:  time.Minute,
 	}
 	if err := processor.Process(context.Background(), retry); err == nil {
@@ -138,6 +172,9 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T)
 	if retry.nacked != time.Minute {
 		t.Fatalf("retry delay = %s", retry.nacked)
 	}
+	if admission.released != 1 {
+		t.Fatalf("released slots = %d", admission.released)
+	}
 
 	poison := &fakeMessage{data: []byte("not-json")}
 	if err := processor.Process(context.Background(), poison); err == nil {
@@ -148,6 +185,24 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T)
 	}
 }
 
+func TestProcessorLeavesAssignmentPendingWhenBackendPoolIsFull(t *testing.T) {
+	message := &fakeMessage{data: encodedAssignment(t)}
+	accepter := &fakeAccepter{accepted: true}
+	processor := Processor{
+		TrustDomain: "ddupan.top",
+		Accepter:    accepter,
+		Claims:      &fakeClaims{},
+		Admission:   &fakeAdmission{},
+		RetryDelay:  3 * time.Second,
+	}
+	if err := processor.Process(context.Background(), message); err != nil {
+		t.Fatal(err)
+	}
+	if message.acked != 0 || message.nacked != 3*time.Second {
+		t.Fatalf("message = %#v", message)
+	}
+}
+
 type fakeConsumerManager struct{ config jetstream.ConsumerConfig }
 
 func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string, config jetstream.ConsumerConfig) (jetstream.Consumer, error) {
@@ -160,7 +215,7 @@ func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
 	if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
 		t.Fatal(err)
 	}
-	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 {
+	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
 		t.Fatalf("config = %#v", manager.config)
 	}
 }

internal/backendpool/pool.go

@@ -0,0 +1,50 @@
+// Package backendpool manages runtime capacity independently for each executor backend.
+package backendpool
+
+import "sync"
+
+type Pool struct {
+	mu       sync.Mutex
+	capacity int
+	active   map[string]struct{}
+}
+
+func New(capacity int) *Pool {
+	return &Pool{capacity: capacity, active: make(map[string]struct{})}
+}
+
+// Acquire reserves a backend slot without blocking. Redelivery of the same
+// assignment is idempotent and succeeds even while the pool is full.
+func (p *Pool) Acquire(assignmentID string) bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if _, exists := p.active[assignmentID]; exists {
+		return true
+	}
+	if assignmentID == "" || len(p.active) >= p.capacity {
+		return false
+	}
+	p.active[assignmentID] = struct{}{}
+	return true
+}
+
+func (p *Pool) Restore(assignmentID string) {
+	if assignmentID == "" {
+		return
+	}
+	p.mu.Lock()
+	p.active[assignmentID] = struct{}{}
+	p.mu.Unlock()
+}
+
+func (p *Pool) Release(assignmentID string) {
+	p.mu.Lock()
+	delete(p.active, assignmentID)
+	p.mu.Unlock()
+}
+
+func (p *Pool) Active() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.active)
+}

internal/backendpool/pool_test.go

@@ -0,0 +1,26 @@
+package backendpool
+
+import "testing"
+
+func TestPoolSeparatesRuntimeCapacityFromDeliveries(t *testing.T) {
+	pool := New(2)
+	if !pool.Acquire("one") || !pool.Acquire("two") || pool.Acquire("three") {
+		t.Fatal("capacity was not enforced")
+	}
+	if !pool.Acquire("one") {
+		t.Fatal("redelivery must be idempotent")
+	}
+	pool.Release("one")
+	if !pool.Acquire("three") || pool.Active() != 2 {
+		t.Fatalf("active=%d", pool.Active())
+	}
+}
+
+func TestRestoreMayTemporarilyExceedReducedCapacity(t *testing.T) {
+	pool := New(1)
+	pool.Restore("one")
+	pool.Restore("two")
+	if pool.Active() != 2 || pool.Acquire("three") {
+		t.Fatalf("active=%d", pool.Active())
+	}
+}

internal/opensandboxbackend/backend.go

@@ -3,9 +3,13 @@ package opensandboxbackend
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
+	"sort"
+	"strings"
 	"time"
 
 	opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
@@ -17,6 +21,8 @@ import (
 
 const assignmentMetadata = "ci.ddupan.top/assignment-id"
 const terminalMetadata = "ci.ddupan.top/terminal"
+const annotationsMetadataPrefix = "ci.ddupan.top/annotations-"
+const metadataValueLimit = 63
 
 type Lifecycle interface {
 	ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error)
@@ -108,6 +114,34 @@ type Backend struct {
 	Config    Config
 }
 
+func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]taskassignment.Assignment, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
+		Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list recoverable sandboxes: %w", err)
+	}
+	assignments := make([]taskassignment.Assignment, 0, len(result.Items))
+	for _, sandbox := range result.Items {
+		if sandbox.Metadata[terminalMetadata] == "true" {
+			continue
+		}
+		annotations, err := decodeAnnotations(sandbox.Metadata)
+		if err != nil {
+			return nil, fmt.Errorf("decode sandbox %s annotations: %w", sandbox.ID, err)
+		}
+		assignment, err := taskassignment.FromMetadata(sandbox.Metadata, annotations, trustDomain)
+		if err != nil {
+			return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
+		}
+		assignments = append(assignments, assignment)
+	}
+	return assignments, nil
+}
+
 func NewLifecycleClient(baseURL, apiKey string, client *http.Client) *opensandbox.LifecycleClient {
 	if client != nil {
 		return opensandbox.NewLifecycleClient(baseURL, apiKey, opensandbox.WithHTTPClient(client))
@@ -145,8 +179,12 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen
 	for key, value := range launch.Environment {
 		environment[key] = value
 	}
-	sandboxMetadata := clone(launch.Metadata.Annotations)
-	for key, value := range launch.Metadata.Labels {
+	sandboxMetadata := clone(launch.Metadata.Labels)
+	annotations, err := encodeAnnotations(launch.Metadata.Annotations)
+	if err != nil {
+		return nil, fmt.Errorf("encode sandbox annotations: %w", err)
+	}
+	for key, value := range annotations {
 		sandboxMetadata[key] = value
 	}
 	request := opensandbox.CreateSandboxRequest{
@@ -170,7 +208,11 @@ func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor
 	if err != nil {
 		return fmt.Errorf("verify sandbox identity metadata: %w", err)
 	}
-	if sandbox.Metadata["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
+	annotations, err := decodeAnnotations(sandbox.Metadata)
+	if err != nil {
+		return fmt.Errorf("decode sandbox identity metadata: %w", err)
+	}
+	if annotations["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
 		return fmt.Errorf("sandbox %s has inconsistent SPIFFE identity metadata", executor.Name)
 	}
 	return nil
@@ -219,3 +261,50 @@ func clone(source map[string]string) map[string]string {
 	}
 	return result
 }
+
+// OpenSandbox metadata follows Kubernetes label-value constraints, unlike Pod
+// annotations. Store the annotation map as deterministic URL-safe base64
+// chunks so repository paths and SPIFFE IDs remain lossless and recoverable.
+func encodeAnnotations(annotations map[string]string) (map[string]string, error) {
+	data, err := json.Marshal(annotations)
+	if err != nil {
+		return nil, err
+	}
+	encoded := base64.RawURLEncoding.EncodeToString(data)
+	result := make(map[string]string, (len(encoded)+metadataValueLimit-1)/metadataValueLimit)
+	for index := 0; len(encoded) > 0; index++ {
+		length := min(metadataValueLimit, len(encoded))
+		result[fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index)] = encoded[:length]
+		encoded = encoded[length:]
+	}
+	return result, nil
+}
+
+func decodeAnnotations(metadata map[string]string) (map[string]string, error) {
+	keys := make([]string, 0)
+	for key := range metadata {
+		if strings.HasPrefix(key, annotationsMetadataPrefix) {
+			keys = append(keys, key)
+		}
+	}
+	if len(keys) == 0 {
+		return nil, errors.New("sandbox annotation metadata is missing")
+	}
+	sort.Strings(keys)
+	var encoded strings.Builder
+	for index, key := range keys {
+		if key != fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index) {
+			return nil, errors.New("sandbox annotation metadata chunks are incomplete")
+		}
+		encoded.WriteString(metadata[key])
+	}
+	data, err := base64.RawURLEncoding.DecodeString(encoded.String())
+	if err != nil {
+		return nil, err
+	}
+	var annotations map[string]string
+	if err := json.Unmarshal(data, &annotations); err != nil {
+		return nil, err
+	}
+	return annotations, nil
+}

internal/opensandboxbackend/backend_test.go

@@ -31,7 +31,8 @@ func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.S
 			return &item, nil
 		}
 	}
-	return &opensandbox.SandboxInfo{ID: id, Metadata: map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID}}, nil
+	metadata, _ := encodeAnnotations(map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID})
+	return &opensandbox.SandboxInfo{ID: id, Metadata: metadata}, nil
 }
 func (f *fakeLifecycle) PatchSandboxMetadata(_ context.Context, id string, patch opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error) {
 	for index := range f.items {
@@ -89,12 +90,38 @@ func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
+	if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Metadata["ci.ddupan.top/runner"] != "true" || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
 		t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor)
 	}
 	if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" {
 		t.Fatalf("environment = %#v", lifecycle.created.Env)
 	}
+	annotations, err := decodeAnnotations(lifecycle.created.Metadata)
+	if err != nil || annotations["ci.ddupan.top/repository"] != "owner/repo" || annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
+		t.Fatalf("annotations=%#v err=%v", annotations, err)
+	}
+	for _, value := range lifecycle.created.Metadata {
+		if len(value) > metadataValueLimit {
+			t.Fatalf("metadata value exceeds %d characters: %q", metadataValueLimit, value)
+		}
+	}
+}
+
+func TestAnnotationMetadataRoundTripPreservesSlashValues(t *testing.T) {
+	want := taskworker.BackendMetadata(assignment()).Annotations
+	encoded, err := encodeAnnotations(want)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := decodeAnnotations(encoded)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for key, value := range want {
+		if got[key] != value {
+			t.Fatalf("%s=%q, want %q", key, got[key], value)
+		}
+	}
 }
 
 func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) {

internal/podbackend/backend.go

@@ -116,6 +116,28 @@ type Backend struct {
 	Config Config
 }
 
+func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assignment, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod")
+	if err != nil {
+		return nil, fmt.Errorf("list recoverable assignment Pods: %w", err)
+	}
+	assignments := make([]taskassignment.Assignment, 0, len(pods))
+	for _, pod := range pods {
+		if pod.Labels[terminalLabel] == "true" {
+			continue
+		}
+		assignment, err := taskassignment.FromMetadata(pod.Labels, pod.Annotations, b.Config.TrustDomain)
+		if err != nil {
+			return nil, fmt.Errorf("recover Pod %s: %w", pod.Name, err)
+		}
+		assignments = append(assignments, assignment)
+	}
+	return assignments, nil
+}
+
 func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
 	if err := b.validate(); err != nil {
 		return nil, err

internal/podbackend/client.go

@@ -8,6 +8,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -76,17 +77,26 @@ func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, erro
 			Containers: []corev1.Container{{
 				Name: "executor", Image: manifest.Image, Args: manifest.Args, Env: environment,
 				SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)},
-				VolumeMounts: []corev1.VolumeMount{{
-					Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true,
+				VolumeMounts: []corev1.VolumeMount{
+					{Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true},
+					{Name: "docker-data", MountPath: "/var/lib/docker"},
+				},
 			}},
-			}},
-			Volumes: []corev1.Volume{{
+			Volumes: []corev1.Volume{
+				{
 					Name: "spire-agent-socket",
 					VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
 						Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
 					}},
+				},
+				{
+					Name: "docker-data",
+					VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{
+						SizeLimit: resourceQuantity("20Gi"),
 					}},
 				},
+			},
+		},
 	}
 	created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
 	if err != nil {
@@ -169,6 +179,11 @@ func podFromKubernetes(pod corev1.Pod) Pod {
 
 func boolPointer(value bool) *bool { return &value }
 
+func resourceQuantity(value string) *resource.Quantity {
+	quantity := resource.MustParse(value)
+	return &quantity
+}
+
 func stringMap(values map[string]string) map[string]any {
 	result := make(map[string]any, len(values))
 	for key, value := range values {

internal/podbackend/client_test.go

@@ -37,6 +37,12 @@ func TestClientPodLifecycleUsesTypedClient(t *testing.T) {
 	if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" {
 		t.Fatalf("environment = %#v", got)
 	}
+	if got := pod.Spec.Containers[0].VolumeMounts; len(got) != 2 || got[1].Name != "docker-data" || got[1].MountPath != "/var/lib/docker" {
+		t.Fatalf("volume mounts = %#v", got)
+	}
+	if got := pod.Spec.Volumes; len(got) != 2 || got[1].EmptyDir == nil || got[1].EmptyDir.SizeLimit == nil || got[1].EmptyDir.SizeLimit.String() != "20Gi" {
+		t.Fatalf("volumes = %#v", got)
+	}
 	if err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil {
 		t.Fatal(err)
 	}

internal/runnerfacade/facade_test.go

@@ -225,3 +225,15 @@ func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) {
 		t.Fatal("capability scope is invalid")
 	}
 }
+
+func TestRegistryRecoversClaimedAssignment(t *testing.T) {
+	registry := NewRegistry()
+	assignment := facadeAssignment(t)
+	if err := registry.RecoverClaimed(assignment); err != nil {
+		t.Fatal(err)
+	}
+	resolved, err := registry.Resolve(assignment.ID, assignment.Identity.SPIFFEID)
+	if err != nil || resolved.Task.GetId() != assignment.Task.GetId() {
+		t.Fatalf("resolved=%#v err=%v", resolved, err)
+	}
+}

internal/runnerfacade/registry.go

@@ -90,3 +90,21 @@ func (r *Registry) Remove(assignmentID string) {
 	defer r.mu.Unlock()
 	delete(r.claims, assignmentID)
 }
+
+// RecoverClaimed restores authorization for an executor that already claimed
+// its task before the controller restarted.
+func (r *Registry) RecoverClaimed(assignment taskassignment.Assignment) error {
+	ready, err := r.Offer(assignment)
+	if err != nil {
+		return err
+	}
+	r.mu.Lock()
+	entry := r.claims[assignment.ID]
+	if !entry.claimed {
+		entry.claimed = true
+		close(entry.ready)
+	}
+	r.mu.Unlock()
+	<-ready
+	return nil
+}

internal/taskassignment/assignment.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"strconv"
 
 	"gitea.dev/actionslib/pkg/model"
 	runnerv1 "gitea.dev/actionslib/runner/v1"
@@ -32,6 +33,32 @@ type Assignment struct {
 	Identity taskidentity.Identity
 }
 
+// FromMetadata reconstructs the minimal assignment needed to authorize an
+// already-running executor after a controller restart. Backend metadata was
+// originally derived from the trusted Gitea task and is validated again here.
+func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) {
+	taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64)
+	if err != nil || taskID < 1 {
+		return Assignment{}, errors.New("backend metadata has invalid task ID")
+	}
+	backend := Backend(labels["ci.ddupan.top/backend"])
+	if backend != BackendPod && backend != BackendVM {
+		return Assignment{}, errors.New("backend metadata has invalid backend")
+	}
+	id := labels["ci.ddupan.top/assignment-id"]
+	if id != fmt.Sprintf("gitea-task-%d", taskID) {
+		return Assignment{}, errors.New("backend metadata assignment ID does not match task ID")
+	}
+	identity, err := taskidentity.FromMetadata(
+		annotations["ci.ddupan.top/repository"], annotations["ci.ddupan.top/job-key"],
+		annotations["ci.ddupan.top/spiffe-id"], trustDomain,
+	)
+	if err != nil {
+		return Assignment{}, err
+	}
+	return Assignment{ID: id, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
+}
+
 type envelope struct {
 	Version  int                   `json:"version"`
 	ID       string                `json:"id"`
@@ -75,7 +102,7 @@ func backendFromTask(task *runnerv1.Task) (Backend, error) {
 		return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
 	}
 	hasPod := slices.Contains(labels, string(BackendPod))
-	hasVM := slices.Contains(labels, string(BackendVM))
+	hasVM := slices.Contains(labels, string(BackendVM)) || slices.Contains(labels, "vm-dev")
 	if hasPod == hasVM {
 		return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
 	}

internal/taskassignment/assignment_test.go

@@ -28,6 +28,7 @@ func TestNewSelectsBackendFromRunsOn(t *testing.T) {
 	}{
 		{"[self-hosted, pod]", BackendPod},
 		{"[self-hosted, vm]", BackendVM},
+		{"[self-hosted, vm-dev]", BackendVM},
 	} {
 		assignment, err := New(task(t, test.labels), "ddupan.top")
 		if err != nil {
@@ -73,3 +74,18 @@ func TestAssignmentWireRoundTripAndValidation(t *testing.T) {
 		t.Fatal("expected tampered backend to fail")
 	}
 }
+
+func TestFromMetadataRecoversMinimalAssignment(t *testing.T) {
+	assignment, err := FromMetadata(map[string]string{
+		"ci.ddupan.top/assignment-id": "gitea-task-42",
+		"ci.ddupan.top/task-id":       "42",
+		"ci.ddupan.top/backend":       "vm",
+	}, map[string]string{
+		"ci.ddupan.top/repository": "owner/repo",
+		"ci.ddupan.top/job-key":    "publish",
+		"ci.ddupan.top/spiffe-id":  "spiffe://ddupan.top/ci/owner/repo/publish",
+	}, "ddupan.top")
+	if err != nil || assignment.Task.GetId() != 42 || assignment.Backend != BackendVM {
+		t.Fatalf("assignment=%#v err=%v", assignment, err)
+	}
+}

internal/taskidentity/identity.go

@@ -23,6 +23,21 @@ type Identity struct {
 	SPIFFEID   string
 }
 
+// FromMetadata validates identity fields recovered from backend-owned state.
+func FromMetadata(repository, task, spiffeID, trustDomain string) (Identity, error) {
+	parts := strings.Split(repository, "/")
+	if len(parts) != 2 || parts[0] == "" || parts[1] == "" || !safeTaskKey.MatchString(task) {
+		return Identity{}, errors.New("invalid recovered repository or task identity")
+	}
+	expected := "spiffe://" + trustDomain + "/ci/" + strings.Join([]string{
+		sanitize(parts[0]), sanitize(parts[1]), task,
+	}, "/")
+	if spiffeID != expected {
+		return Identity{}, fmt.Errorf("recovered SPIFFE ID %q does not match %q", spiffeID, expected)
+	}
+	return Identity{Repository: repository, Task: task, SPIFFEID: spiffeID}, nil
+}
+
 // FromTask derives the repository/task SPIFFE ID from Gitea's trusted task
 // context. Workflow input never supplies or overrides the resulting ID.
 func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) {

internal/taskscheduler/gate.go

@@ -1,31 +0,0 @@
-package taskscheduler
-
-import "context"
-
-// SingleFlightGate keeps at most one fetched task in flight. Release is
-// idempotent so repeated terminal updates cannot increase capacity.
-type SingleFlightGate struct {
-	token chan struct{}
-}
-
-func NewSingleFlightGate() *SingleFlightGate {
-	gate := &SingleFlightGate{token: make(chan struct{}, 1)}
-	gate.token <- struct{}{}
-	return gate
-}
-
-func (g *SingleFlightGate) Acquire(ctx context.Context) error {
-	select {
-	case <-g.token:
-		return nil
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-}
-
-func (g *SingleFlightGate) Release() {
-	select {
-	case g.token <- struct{}{}:
-	default:
-	}
-}

internal/taskscheduler/gate_test.go

@@ -1,29 +0,0 @@
-package taskscheduler
-
-import (
-	"context"
-	"testing"
-	"time"
-)
-
-func TestSingleFlightGateBlocksUntilTerminalRelease(t *testing.T) {
-	gate := NewSingleFlightGate()
-	if err := gate.Acquire(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	blocked, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
-	defer cancel()
-	if err := gate.Acquire(blocked); err == nil {
-		t.Fatal("second task acquired capacity before release")
-	}
-	gate.Release()
-	gate.Release()
-	if err := gate.Acquire(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	blockedAgain, cancelAgain := context.WithTimeout(context.Background(), 20*time.Millisecond)
-	defer cancelAgain()
-	if err := gate.Acquire(blockedAgain); err == nil {
-		t.Fatal("duplicate terminal release increased capacity")
-	}
-}

internal/taskscheduler/poller.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync/atomic"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 )
 
@@ -19,14 +22,15 @@ type PollerConfig struct {
 	Labels       []string
 	EmptyBackoff time.Duration
 	ErrorBackoff time.Duration
+	Capacity     int
 }
 
-// Poller is the scheduler component. Once Gitea assigns a task, it never
-// fetches another one until the current assignment is durably dispatched.
+// Poller is the scheduler component. Each fetcher keeps its assigned task
+// until that assignment is durably dispatched; all fetchers share one runner
+// declaration and a monotonic tasks version.
 type Poller struct {
 	Client    PollClient
 	Scheduler *Scheduler
-	Gate      *SingleFlightGate
 	Config    PollerConfig
 	OnError   func(error)
 }
@@ -50,16 +54,21 @@ func (p Poller) Run(ctx context.Context) error {
 	if errorBackoff <= 0 {
 		errorBackoff = 5 * time.Second
 	}
-	var tasksVersion int64
-	haveLease := false
+	capacity := p.Config.Capacity
+	if capacity < 1 {
+		capacity = 1
+	}
+	var tasksVersion atomic.Int64
+	group, groupContext := errgroup.WithContext(ctx)
+	for range capacity {
+		group.Go(func() error { return p.runFetcher(groupContext, &tasksVersion, emptyBackoff, errorBackoff) })
+	}
+	return group.Wait()
+}
+
+func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, emptyBackoff, errorBackoff time.Duration) error {
 	for {
-		if p.Gate != nil && !haveLease {
-			if err := p.Gate.Acquire(ctx); err != nil {
-				return nil
-			}
-			haveLease = true
-		}
-		response, err := p.Client.FetchTask(ctx, tasksVersion)
+		response, err := p.Client.FetchTask(ctx, tasksVersion.Load())
 		if err != nil {
 			if ctx.Err() != nil {
 				return nil
@@ -77,13 +86,9 @@ func (p Poller) Run(ctx context.Context) error {
 			}
 			continue
 		}
-		tasksVersion = response.GetTasksVersion()
+		storeMaximum(tasksVersion, response.GetTasksVersion())
 		task := response.GetTask()
 		if task == nil {
-			if p.Gate != nil {
-				p.Gate.Release()
-				haveLease = false
-			}
 			if !wait(ctx, emptyBackoff) {
 				return nil
 			}
@@ -91,7 +96,6 @@ func (p Poller) Run(ctx context.Context) error {
 		}
 		for {
 			if err := p.Scheduler.Run(ctx, task); err == nil {
-				haveLease = false
 				break
 			} else {
 				if ctx.Err() != nil {
@@ -106,6 +110,14 @@ func (p Poller) Run(ctx context.Context) error {
 	}
 }
 
+func storeMaximum(value *atomic.Int64, candidate int64) {
+	for current := value.Load(); candidate > current; current = value.Load() {
+		if value.CompareAndSwap(current, candidate) {
+			return
+		}
+	}
+}
+
 func (p Poller) report(err error) {
 	if p.OnError != nil {
 		p.OnError(err)

internal/taskscheduler/poller_test.go

@@ -115,3 +115,52 @@ func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) {
 		t.Fatalf("dispatches=%d fetches-before-dispatch=%d declares=%d", dispatcher.calls, dispatcher.fetchesAtSuccess, client.declared)
 	}
 }
+
+type blockingPollClient struct {
+	mu       sync.Mutex
+	declared int
+	started  chan struct{}
+}
+
+func (c *blockingPollClient) Declare(context.Context, string, []string) error {
+	c.mu.Lock()
+	c.declared++
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *blockingPollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) {
+	c.started <- struct{}{}
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func TestPollerStartsConfiguredNumberOfFetchersAfterOneDeclare(t *testing.T) {
+	client := &blockingPollClient{started: make(chan struct{}, 3)}
+	poller := Poller{
+		Client:    client,
+		Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: &retryDispatcher{done: make(chan struct{})}},
+		Config: PollerConfig{
+			Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"}, Capacity: 3,
+		},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	finished := make(chan error, 1)
+	go func() { finished <- poller.Run(ctx) }()
+	for range 3 {
+		select {
+		case <-client.started:
+		case <-time.After(time.Second):
+			t.Fatal("configured fetchers did not start")
+		}
+	}
+	cancel()
+	if err := <-finished; err != nil {
+		t.Fatal(err)
+	}
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if client.declared != 1 {
+		t.Fatalf("declares = %d", client.declared)
+	}
+}

internal/taskworker/worker.go

@@ -71,6 +71,28 @@ type Worker struct {
 	Backend   Backend
 	Tasks     TaskState
 	Bootstrap Bootstrap
+	OnEvent   func(Event)
+}
+
+// Event describes a backend lifecycle transition without exposing launch
+// environment values or other credentials.
+type Event struct {
+	Name         string
+	AssignmentID string
+	Executor     string
+	Phase        Phase
+}
+
+func (w Worker) event(name string, assignment taskassignment.Assignment, executor *Executor) {
+	if w.OnEvent == nil {
+		return
+	}
+	event := Event{Name: name, AssignmentID: assignment.ID}
+	if executor != nil {
+		event.Executor = executor.Name
+		event.Phase = executor.Phase
+	}
+	w.OnEvent(event)
 }
 
 // Accept completes the durable handoff from JetStream to the backend. Once it
@@ -88,6 +110,7 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
 		return false, err
 	}
 	if executor == nil {
+		w.event("executor_absent", assignment, nil)
 		launch, launchErr := w.launchSpec(assignment)
 		if launchErr != nil {
 			return false, launchErr
@@ -96,13 +119,18 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
 		if err != nil {
 			return false, err
 		}
+		w.event("executor_created", assignment, executor)
+	} else {
+		w.event("executor_found", assignment, executor)
 	}
 	if executor.IdentityTarget == "" {
+		w.event("identity_target_pending", assignment, executor)
 		return false, nil
 	}
 	if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
 		return false, err
 	}
+	w.event("identity_bound", assignment, executor)
 	return true, nil
 }
 
@@ -178,6 +206,7 @@ func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, er
 func BackendMetadata(assignment taskassignment.Assignment) Metadata {
 	return Metadata{
 		Labels: map[string]string{
+			"ci.ddupan.top/runner":        "true",
 			"ci.ddupan.top/assignment-id": assignment.ID,
 			"ci.ddupan.top/task-id":       strconv.FormatInt(assignment.Task.GetId(), 10),
 			"ci.ddupan.top/backend":       string(assignment.Backend),

internal/taskworker/worker_test.go

@@ -79,7 +79,8 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
 
 func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
 	backend := &fakeBackend{}
-	worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
+	var events []Event
+	worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}, OnEvent: func(event Event) { events = append(events, event) }}
 
 	accepted, err := worker.Accept(context.Background(), assignment())
 	if err != nil || !accepted {
@@ -88,6 +89,18 @@ func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
 	if backend.created != 1 || backend.bound != 1 || backend.deleted != 0 {
 		t.Fatalf("created=%d bound=%d deleted=%d", backend.created, backend.bound, backend.deleted)
 	}
+	want := []string{"executor_absent", "executor_created", "identity_bound"}
+	if len(events) != len(want) {
+		t.Fatalf("events = %#v", events)
+	}
+	for index := range want {
+		if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" {
+			t.Fatalf("event[%d] = %#v", index, events[index])
+		}
+	}
+	if events[1].Executor != "executor" || events[1].Phase != PhaseRunning {
+		t.Fatalf("created event = %#v", events[1])
+	}
 }
 
 func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) {

scripts/gitea-job-started

@@ -10,6 +10,7 @@ while [ "$(date +%s)" -lt "$deadline" ]; do
         -audience ci-job-ready \
         -socketPath "$socket" \
         >/dev/null 2>&1; then
+        /usr/local/libexec/setup-job-docker
         exit 0
     fi
     sleep 1

scripts/setup-job-docker

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if docker info >/dev/null 2>&1; then
+  printf '%s\n' 'job Docker daemon is already ready'
+  exit 0
+fi
+
+storage_size=${DOCKER_DATA_SIZE:-20G}
+storage_driver=${DOCKER_STORAGE_DRIVER:-overlay2}
+wait_seconds=${DOCKER_START_WAIT_SECONDS:-60}
+
+sudo install -d /var/lib/docker
+if ! mountpoint --quiet /var/lib/docker; then
+  printf '%s\n' "preparing ${storage_size} loop-backed Docker storage"
+  if [[ ! -e /dev/loop-control ]]; then
+    sudo mknod /dev/loop-control c 10 237
+  fi
+  for minor in {0..7}; do
+    if [[ ! -e "/dev/loop${minor}" ]]; then
+      sudo mknod "/dev/loop${minor}" b 7 "$minor"
+    fi
+  done
+  sudo truncate -s "$storage_size" /tmp/docker-data.img
+  sudo mkfs.ext4 -F /tmp/docker-data.img
+  loop_device=$(sudo losetup --find --show /tmp/docker-data.img)
+  sudo mount "$loop_device" /var/lib/docker
+else
+  printf '%s\n' 'using mounted Docker storage at /var/lib/docker'
+fi
+
+if [[ -f /sys/fs/cgroup/cgroup.controllers ]]; then
+  sudo sh -c '
+    mkdir -p /sys/fs/cgroup/init
+    while read -r pid; do
+      printf "%s\n" "$pid" \
+        >/sys/fs/cgroup/init/cgroup.procs 2>/dev/null || true
+    done </sys/fs/cgroup/cgroup.procs
+    sed -e "s/ / +/g" -e "s/^/+/" \
+      /sys/fs/cgroup/cgroup.controllers \
+      >/sys/fs/cgroup/cgroup.subtree_control
+  '
+fi
+
+sudo sh -c 'exec nohup dockerd "$@" >/tmp/dockerd.log 2>&1' sh \
+  --host=unix:///var/run/docker.sock \
+  --storage-driver="$storage_driver" &
+printf '%s\n' 'waiting for job Docker daemon'
+for ((attempt = 0; attempt < wait_seconds; attempt++)); do
+  if docker info >/dev/null 2>&1; then
+    findmnt /var/lib/docker
+    docker info --format \
+      '{{json .ServerVersion}} {{json .Driver}} {{json .CgroupVersion}}'
+    exit 0
+  fi
+  sleep 1
+done
+
+cat /tmp/dockerd.log
+exit 1