ci: 为 Kata nested kubelet 补充 kmsg 设备

.gitea/workflows/pod-smoke.yml

@@ -16,4 +16,4 @@ jobs:
             -audience ci-smoke \
             -socketPath /run/spire/agent-sockets/spire-agent.sock \
             >/dev/null
-          test "$(id -u)" = 0
+          test "$(id -u)" = 2000

.gitea/workflows/publish-images.yml

@@ -8,8 +8,12 @@ on:
       - '.gitea/workflows/publish-images.yml'
       - 'config/**'
       - 'container/**'
+      - 'cmd/**'
+      - 'internal/**'
       - 'scripts/**'
       - 'src/**'
+      - 'go.mod'
+      - 'go.sum'
       - 'pyproject.toml'
       - 'README.md'
   workflow_dispatch:
@@ -34,6 +38,8 @@ jobs:
         shell: bash
         run: |
           set -euo pipefail
+          go test ./...
+          go vet ./...
           python3 -m pip install --break-system-packages -e '.[test]'
           pytest -q
           python3 -m compileall -q src tests

.gitea/workflows/test.yml

@@ -6,8 +6,18 @@ on:
   pull_request:
 
 jobs:
+  go:
+    runs-on: [self-hosted, pod]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v6
+        with:
+          go-version-file: go.mod
+      - run: go test ./...
+      - run: go vet ./...
+
   python:
-    runs-on: self-hosted
+    runs-on: [self-hosted, pod]
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -18,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,31 +1,154 @@
+---
 name: VM kind smoke
 
+# yamllint disable-line rule:truthy
 on:
   workflow_dispatch:
 
 jobs:
   kind:
-    runs-on: [self-hosted, vm]
+    runs-on: [self-hosted, vm-dev]
     steps:
+      - name: Start job-local Docker
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [[ ! -e /dev/kmsg ]]; then
+            sudo mknod /dev/kmsg c 1 11
+          fi
+          sudo install -d /var/lib/docker
+          sudo truncate -s 20G /tmp/docker-data.img
+          sudo mkfs.ext4 -F /tmp/docker-data.img
+          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
+          loop_device=$(sudo losetup --find --show /tmp/docker-data.img)
+          sudo mount "$loop_device" /var/lib/docker
+          findmnt /var/lib/docker
+
+          # A nested systemd needs a domain cgroup namespace. This is the
+          # cgroup v2 nesting initialization performed by the official DinD
+          # entrypoint, kept here because the shared runner image deliberately
+          # does not carry a second DinD-specific entrypoint.
+          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 nohup dockerd \
+            --host=unix:///var/run/docker.sock \
+            --storage-driver=overlay2 \
+            >/tmp/dockerd.log 2>&1 &
+          for _ in {1..60}; do
+            if docker info >/dev/null 2>&1; then
+              exit 0
+            fi
+            sleep 1
+          done
+          cat /tmp/dockerd.log
+          exit 1
+
       - name: Verify Docker
-        run: docker info
+        shell: bash
+        run: |
+          set -euo pipefail
+          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
+          /tmp/kind create cluster --name smoke --wait 180s --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-dev]
+    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,11 +15,19 @@ 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 修复过程中继续涌入。
 
-- `controller`：接收 Gitea `workflow_job` webhook，将任务持久化到 NATS JetStream；仅在
-  显式启用 VM consumer 时调用 OpenSandbox Lifecycle
-  API，从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。
+目标 Go controller 组件：
+
+- `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment
+  持久化到 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
+  的 assignment claim registry 仍是进程内状态；支持安全拆分前进程会明确拒绝该配置。
 - `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行
   Cloud Hypervisor，退出后完整清理。
 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral
@@ -27,15 +35,16 @@ runs-on: [self-hosted, vm]
 - `opensandbox-identity`：在 sandbox 集群按实际 Pod UID 创建并清理临时 SPIFFE
   entry；不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
   [`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
-- `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用
+- Pod executor：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用
   host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
   支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。
 - `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)。
 
 ## 开发
@@ -45,8 +54,33 @@ python -m venv .venv
 . .venv/bin/activate
 pip install -e '.[test]'
 pytest
+
+go test ./...
+go vet ./...
 ```
 
+## Go controller 首次集成配置
+
+controller 默认执行 `controller` 子命令，runner 镜像执行 `executor` 子命令。所有长期
+credential 都从挂载文件读取，不接受明文环境变量：
+
+- `GITEA_RUNNER_UUID_FILE`、`GITEA_RUNNER_TOKEN_FILE`：scheduler 的常驻 RunnerService
+  registration；该 credential 不下发给 executor。
+- `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`：分别使用现有最小权限的
+  `ci-producer` publish 连接和 `ci-worker` pull/ACK 连接，controller 不合并权限。
+- `RUNNER_FACADE_CAPABILITY_KEY_FILE`：至少 32 字节的 controller HMAC key。
+- `OPENSANDBOX_API_KEY_FILE`：仅启用 `vm-worker` 时读取。
+
+必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`（应使用 digest）、`SPIRE_AGENT_ID`、
+`RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认
+`COMPONENTS=all`、Pod 并发 4、VM 并发 1；首次 smoke test 应显式设为
+`COMPONENTS=scheduler,pod-worker`，先验证 Pod 链路，避免同时消耗 VM 容量。
+
+Pod task 的 terminal update 被 Gitea 接受后，controller 会在 Pod 上持久写入
+`ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经
+进入 `Succeeded` 或 `Failed` phase 的 Pod 及其同名 `ClusterStaticEntry`；controller
+重启不影响清理恢复，上报终态前失败的 Pod 也不会被误删。
+
 ## 安全边界
 
 - OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。

cmd/gitea-dynamic-runner/controller.go

@@ -0,0 +1,405 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+
+	"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"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/podbackend"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerbootstrap"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
+)
+
+type runComponent func(context.Context) error
+
+func (function runComponent) Run(ctx context.Context) error { return function(ctx) }
+
+type controllerConfig struct {
+	Components                                                          controller.Selection
+	TrustDomain, WorkloadAPIAddr                                        string
+	GiteaURL, GiteaUUID, GiteaToken                                     string
+	NATSURL, NATSProducerUser, NATSProducerPassword                     string
+	NATSWorkerUser, NATSWorkerPassword, NATSCA, Stream, SubjectBase     string
+	FacadeListen, FacadeURL, FacadeSPIFFEID                             string
+	CapabilityKey                                                       []byte
+	PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
+	SPIREAgentID                                                        string
+	PodExecutorUID, PodCapacity                                         int
+	OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool, VMRunnerLabel   string
+	VMTimeout, VMCapacity                                               int
+}
+
+func runController(ctx context.Context) error {
+	config, err := loadControllerConfig()
+	if err != nil {
+		return err
+	}
+	if !slices.Contains(config.Components, controller.Scheduler) {
+		return errors.New("split worker deployment is not yet safe: scheduler/facade must be enabled with workers")
+	}
+	if !slices.Contains(config.Components, controller.PodWorker) && !slices.Contains(config.Components, controller.VMWorker) {
+		return errors.New("scheduler requires at least one local backend worker")
+	}
+
+	producerConnection, err := connectNATS(config.NATSURL, config.NATSProducerUser, config.NATSProducerPassword, config.NATSCA, "gitea-dynamic-runner-producer")
+	if err != nil {
+		return fmt.Errorf("connect NATS producer: %w", err)
+	}
+	defer producerConnection.Close()
+	producerJS, err := jetstream.New(producerConnection)
+	if err != nil {
+		return fmt.Errorf("open producer JetStream: %w", err)
+	}
+	if _, err := producerJS.Stream(ctx, config.Stream); err != nil {
+		return fmt.Errorf("open assignment stream %s: %w", config.Stream, err)
+	}
+	workerConnection, err := connectNATS(config.NATSURL, config.NATSWorkerUser, config.NATSWorkerPassword, config.NATSCA, "gitea-dynamic-runner-worker")
+	if err != nil {
+		return fmt.Errorf("connect NATS worker: %w", err)
+	}
+	defer workerConnection.Close()
+	workerJS, err := jetstream.New(workerConnection)
+	if err != nil {
+		return fmt.Errorf("open worker JetStream: %w", err)
+	}
+
+	capabilities, err := runnerfacade.NewCapabilities(config.CapabilityKey)
+	if err != nil {
+		return err
+	}
+	registry := runnerfacade.NewRegistry()
+	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)
+	facade := &runnerfacade.Facade{
+		Registry: registry, Capabilities: capabilities, Upstream: giteaClient,
+		OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error {
+			switch assignment.Backend {
+			case taskassignment.BackendPod:
+				if podExecutorBackend == nil {
+					return errors.New("Pod lifecycle backend is not configured")
+				}
+				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")
+				}
+				if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
+					return err
+				}
+				vmPool.Release(assignment.ID)
+			}
+			return nil
+		},
+	}
+	bootstrap := runnerbootstrap.Bootstrap{
+		Capabilities: capabilities, FacadeURL: config.FacadeURL, FacadeSPIFFEID: config.FacadeSPIFFEID,
+		WorkloadAPIAddr: config.WorkloadAPIAddr,
+	}
+
+	labels := []string{"self-hosted"}
+	if slices.Contains(config.Components, controller.PodWorker) {
+		labels = append(labels, string(taskassignment.BackendPod))
+	}
+	if slices.Contains(config.Components, controller.VMWorker) {
+		labels = append(labels, config.VMRunnerLabel)
+	}
+	poller := taskscheduler.Poller{
+		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, 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,
+		WorkloadAPIAddr: config.WorkloadAPIAddr, UpstreamURL: config.GiteaURL,
+	}
+	components := controller.Registry{
+		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 runSchedulerLeader(groupContext, kubernetesClient, config.PodNamespace, leaderIdentity, poller.Run)
+			})
+			return group.Wait()
+		}),
+	}
+
+	if slices.Contains(config.Components, controller.PodWorker) {
+		client, err := podbackend.NewInClusterClient()
+		if err != nil {
+			return err
+		}
+		backend := podbackend.Backend{API: client, Config: podbackend.Config{
+			Namespace: config.PodNamespace, Image: config.PodImage, ServiceAccount: config.PodServiceAccount,
+			ExecutorArgs: []string{"executor"}, TrustDomain: config.TrustDomain,
+			SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass,
+			SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
+		}}
+		podExecutorBackend = &backend
+		assignments, err := backend.RecoverAssignments(ctx)
+		if err != nil {
+			return 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) })
+			group.Go(func() error { return lifecycle.Run(groupContext) })
+			return group.Wait()
+		})
+	}
+	if slices.Contains(config.Components, controller.VMWorker) {
+		lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second})
+		backend := opensandboxbackend.Backend{Lifecycle: lifecycle, Config: opensandboxbackend.Config{
+			Pool: config.OpenSandboxPool, Timeout: config.VMTimeout,
+			Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"},
+			Env:        map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
+		}}
+		vmExecutorBackend = &backend
+		assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
+		if err != nil {
+			return 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) })
+			group.Go(func() error { return lifecycleReconciler.Run(groupContext) })
+			return group.Wait()
+		})
+	}
+	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 != "" {
+		options = append(options, nats.RootCAs(caFile))
+	}
+	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, 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, 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 {
+		return controllerConfig{}, err
+	}
+	read := func(name string) (string, error) {
+		path := os.Getenv(name)
+		if path == "" {
+			return "", fmt.Errorf("%s is required", name)
+		}
+		value, err := os.ReadFile(path)
+		if err != nil {
+			return "", fmt.Errorf("read %s: %w", name, err)
+		}
+		return strings.TrimSpace(string(value)), nil
+	}
+	uuid, err := read("GITEA_RUNNER_UUID_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	token, err := read("GITEA_RUNNER_TOKEN_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	producerPassword, err := read("NATS_PRODUCER_PASSWORD_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	workerPassword, err := read("NATS_WORKER_PASSWORD_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	capabilityKey, err := read("RUNNER_FACADE_CAPABILITY_KEY_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	config := controllerConfig{
+		Components: selection, TrustDomain: env("TRUST_DOMAIN", "ddupan.top"), WorkloadAPIAddr: os.Getenv("SPIFFE_ENDPOINT_SOCKET"),
+		GiteaURL: env("GITEA_INSTANCE_URL", "https://git.ddupan.top"), GiteaUUID: uuid, GiteaToken: token,
+		NATSURL:          env("NATS_URL", "tls://nats.ad.ddupan.top:4222"),
+		NATSProducerUser: env("NATS_PRODUCER_USER", "ci-producer"), NATSProducerPassword: producerPassword,
+		NATSWorkerUser: env("NATS_WORKER_USER", "ci-worker"), NATSWorkerPassword: workerPassword,
+		NATSCA: os.Getenv("NATS_CA_FILE"), Stream: env("NATS_STREAM", "CI_RUNNER"), SubjectBase: env("NATS_SUBJECT_BASE", "ci.runner"),
+		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"), 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")
+	}
+	if slices.Contains(selection, controller.PodWorker) && config.PodImage == "" {
+		return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for pod-worker")
+	}
+	if slices.Contains(selection, controller.PodWorker) && config.SPIREAgentID == "" {
+		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")
+		}
+		config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE")
+		if err != nil {
+			return controllerConfig{}, err
+		}
+	}
+	return config, nil
+}
+
+func env(name, fallback string) string {
+	if value := strings.TrimSpace(os.Getenv(name)); value != "" {
+		return value
+	}
+	return fallback
+}
+
+func envInt(name string, fallback int) int {
+	value := strings.TrimSpace(os.Getenv(name))
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil || parsed < 1 {
+		return fallback
+	}
+	return parsed
+}

cmd/gitea-dynamic-runner/controller_test.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
+)
+
+func secretFile(t *testing.T, name, value string) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), name)
+	if err := os.WriteFile(path, []byte(value+"\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+func TestLoadControllerConfigUsesFileSecrets(t *testing.T) {
+	t.Setenv("COMPONENTS", "scheduler,pod-worker")
+	t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "scheduler-uuid"))
+	t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "scheduler-token"))
+	t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
+	t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
+	t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
+	t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
+	t.Setenv("RUNNER_FACADE_URL", "https://gitea-runner-facade.gitea-actions.svc:8443")
+	t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner")
+	t.Setenv("POD_EXECUTOR_IMAGE", "zot.ddupan.top/ci/gitea-runner@sha256:abc")
+	t.Setenv("SPIRE_AGENT_ID", "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid")
+
+	config, err := loadControllerConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker {
+		t.Fatalf("components = %#v", config.Components)
+	}
+	if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" {
+		t.Fatal("file secrets were not loaded")
+	}
+	if string(config.CapabilityKey) != "0123456789abcdef0123456789abcdef" || config.PodExecutorUID != 2000 {
+		t.Fatalf("config = %#v", config)
+	}
+}
+
+func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) {
+	t.Setenv("COMPONENTS", "scheduler,vm-worker")
+	t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "uuid"))
+	t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "token"))
+	t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
+	t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
+	t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
+	t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
+	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

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerbootstrap"
+)
+
+func main() {
+	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
+	if err := run(); err != nil {
+		slog.Error("runner stopped", "error", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	if len(os.Args) > 2 {
+		return errors.New("usage: gitea-dynamic-runner [controller|executor]")
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	command := "controller"
+	if len(os.Args) == 2 {
+		command = os.Args[1]
+	}
+	switch command {
+	case "controller":
+		return runController(ctx)
+	case "executor":
+		config, err := runnerbootstrap.ExecutorConfigFromEnvironment()
+		if err != nil {
+			return err
+		}
+		return runnerbootstrap.RunExecutor(ctx, config)
+	default:
+		return fmt.Errorf("unknown command %q", command)
+	}
+}

container/controller.Dockerfile

@@ -1,17 +1,17 @@
-FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
-
-FROM python:3.12.11-alpine3.22 AS build
+FROM docker.io/library/golang:1.27-alpine@sha256:4cb7ac979db5fcc41cae44b2227ba5ab8a51e8807f40d9ba4dee20a0ad960b5b AS build
 WORKDIR /src
-COPY pyproject.toml README.md ./
-COPY src ./src
-RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir .
+COPY go.mod go.sum ./
+RUN go mod download
+COPY cmd ./cmd
+COPY internal ./internal
+RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/gitea-dynamic-runner ./cmd/gitea-dynamic-runner
 
 FROM python:3.12.11-alpine3.22
 RUN addgroup -S -g 65532 runner \
     && adduser -S -D -H -u 65532 -G runner runner \
     && install -d -o 65532 -g 65532 /var/run/secrets/kubernetes.io/serviceaccount
-COPY --from=build /venv /venv
-COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
+COPY --from=build /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
 USER 65532:65532
-EXPOSE 8787
-ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"]
+EXPOSE 8443
+ENTRYPOINT ["/usr/local/bin/gitea-dynamic-runner"]
+CMD ["controller"]

container/runner.Dockerfile

@@ -2,6 +2,14 @@ FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766f
 
 FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS runner
 
+FROM docker.io/library/golang:1.27-alpine@sha256:4cb7ac979db5fcc41cae44b2227ba5ab8a51e8807f40d9ba4dee20a0ad960b5b AS controller
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY cmd ./cmd
+COPY internal ./internal
+RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/gitea-dynamic-runner ./cmd/gitea-dynamic-runner
+
 # The runner daemon image is intentionally minimal and does not contain the
 # Node.js runtime required by JavaScript actions such as actions/checkout.
 # Run the daemon in Gitea's Ubuntu workflow image so host-mode jobs and their
@@ -9,13 +17,22 @@ FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a63
 FROM docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
 USER root
 
+RUN groupadd --gid 2000 runner \
+    && useradd --uid 2000 --gid 2000 --groups docker --create-home --shell /bin/bash runner \
+    && printf 'runner ALL=(ALL) NOPASSWD:ALL\n' >/etc/sudoers.d/runner \
+    && chmod 0440 /etc/sudoers.d/runner \
+    && install -d -o 2000 -g 2000 /data
+
 COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
-COPY --from=runner /usr/local/bin/run.sh /usr/local/bin/run.sh
+COPY --from=controller /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
 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
 
 VOLUME ["/data"]
-WORKDIR /
-ENTRYPOINT ["/usr/local/bin/run.sh"]
+ENV HOME=/home/runner
+USER 2000:2000
+WORKDIR /home/runner
+ENTRYPOINT ["/usr/local/bin/gitea-dynamic-runner"]
+CMD ["executor"]

docs/design-principles.md

@@ -21,18 +21,20 @@ runs-on: [self-hosted, vm]
 
 ## 一个 job，一个环境
 
-Controller 根据 Gitea `workflow_job` webhook 创建执行环境。每个 Pod 或 VM 注册一个
-ephemeral runner，只执行一个 job；任务结束后注销 runner，并删除计算环境及其全部
-本地状态。
+Controller 通过 Gitea RunnerService 原子领取具体 task，再把版本化 assignment 交给
+一个一次性 Pod 或 VM。executor 直接执行已经领取的 task，不再注册临时 runner 去
+二次竞争任务；任务结束并回报 Gitea 后删除计算环境及其全部本地状态。
 
 `job_id` 仅用于消息去重、状态追踪、实例关联和失败清理，不进入 workload 身份，也
 不参与资源授权。
 
-Gitea 不保证由某次 `queued` webhook 创建的 runner 一定领取该 webhook 对应的 job。
-因此创建环境时只赋予无业务权限的启动身份。runner 实际领取任务后，controller 根据
-`in_progress` webhook 返回的 `runner_name` 和真实 job 名称绑定业务身份；环境中的
-job-start hook 必须等目标 SVID 可用后才放行 workflow 的第一步。不能依据 queued
-事件提前赋予任务权限。
+assignment ID 只用于消息去重、状态追踪、实例关联和失败清理，不进入 workload 身份，
+也不参与资源授权。worker 通过 assignment ID 从 Kubernetes labels 或 OpenSandbox
+metadata 恢复 executor；JetStream 不保存 executor 生命周期状态。
+
+Pod executor 创建后，controller 使用实际 Pod UID 创建幂等 `ClusterStaticEntry`，将
+SPIFFE ID绑定到该 Pod 的 workload selector。executor 必须等目标 SVID 可用后才执行
+workflow 的第一步。
 
 ## 环境只提供运行边界
 
@@ -54,15 +56,15 @@ kind 等工具由 pipeline 按需安装和启动，而不是由 controller 预
 SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生：
 
 ```text
-spiffe://ddupan.top/ci/<owner>/<repository>/<job-name>
+spiffe://ddupan.top/ci/<owner>/<repository>/<job-key>
 ```
 
 同一种任务在不同运行中使用相同的逻辑 SPIFFE ID；每次运行取得独立、短期的 SVID。
 Pod 与 VM 是可替换的执行实现，因此默认不写入 SPIFFE ID。
 
-job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在发生冲突时
-拒绝创建环境，不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用
-不同的 job 名称；workflow 文件只是编排载体，不进入权限身份。
+job key 必须满足 `[A-Za-z_][A-Za-z0-9_-]*`，展示名称 `name` 不参与身份计算。同一
+仓库内需要不同权限的任务应使用不同的 job key；workflow 文件只是编排载体，不进入
+权限身份。
 
 ## Self-service 与授权边界
 
@@ -86,3 +88,11 @@ OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意
 
 仓库中的 `jwt-broker` 是早期方案的实验实现，在 Pod/VM 动态执行环境完成迁移后不应
 部署。
+
+## 实现依赖原则
+
+基础设施协议优先使用上游维护的成熟客户端，不在 controller 内重复实现认证、连接、
+资源编码或错误语义。Kubernetes 使用 `client-go`，NATS JetStream 使用 `nats.go`，
+Gitea RunnerService 使用 `actionslib`，OpenSandbox Lifecycle API 使用官方 Go SDK；
+SPIFFE Workload API 与 mTLS 使用 `go-spiffe`。自定义代码只保留领域模型、reconcile
+规则及上游客户端未覆盖的最小适配层。

docs/opensandbox-runner.md

@@ -40,15 +40,18 @@ 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 独立运行，需要 Docker 的 workflow 通过 sudo 在
+  privileged executor 内启动 job-local daemon；
+- Kata VM 中 Docker 数据使用 guest 内的 loop-backed ext4，并随 sandbox 一起删除；
 - `self-hosted` 必须是所有 runner labels 的前缀；
 - ephemeral/once runner 完成一项任务后退出。
 
 ## 清理与恢复
 
-controller 监控 Lifecycle 状态，在终止、失败、超时或取消时调用 DELETE。API delete、
-identity entry delete 均接受对象已不存在。controller 重启时，OpenSandbox timeout
-仍是最终回收边界；后续可基于 metadata list 恢复主动监控，但不得为此重新引入消息
-队列。
+Gitea 接受 runner 终态后，facade 先通过 Lifecycle API 将
+`ci.ddupan.top/terminal=true` 持久化到 sandbox metadata，再向 runner 返回成功。VM
+lifecycle reconciler 按该 metadata 查询并调用 DELETE；controller 在标记与删除之间重启
+也能恢复清理。API delete、identity entry delete 均接受对象已不存在，OpenSandbox
+timeout 仍是最终兜底回收边界；生命周期状态不写入消息队列或新的数据库。

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
@@ -18,6 +19,28 @@ dynamic-runner scheduler
                 └──────────────────────► Gitea
 ```
 
+controller 使用单一 Go 二进制；默认在同一进程启用 `scheduler`、`pod-worker` 和
+`vm-worker`，也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界，
+共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。
+
+首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择
+接口，但当前会拒绝“worker 不带 scheduler/facade”的拆分部署：普通 Kubernetes Service
+无法保证 executor 回到持有其 assignment 的 replica。后续拆分必须增加按 assignment
+路由或可重建的 claim 分发，不能新增一套生命周期数据库来掩盖该问题。
+
+executor 直接运行固定版本的官方 Gitea Runner 二进制，不 fork workflow 执行引擎。
+controller 暴露兼容 RunnerService 的 facade：`FetchTask` 只返回已分配 assignment，
+`UpdateTask` 与 `UpdateLog` 转发真实 Gitea。facade 同时验证逻辑 SPIFFE ID、assignment
+ID，以及由 controller 密钥确定性生成的 assignment HMAC capability；该 capability
+只绑定执行实例，不参与 Zot/OpenBao 等业务授权。
+
+executor 为官方 runner 生成与 v3.5.0 schema 一致的一次性 `.runner` 文件，并以
+`daemon --once` 启动。runner 只访问 executor 内的 loopback HTTP proxy；proxy 使用
+`go-spiffe` 从 Workload API 持续取得和轮换 X509-SVID，再以 mTLS 连接 controller
+facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把静态客户端证书写入
+镜像。assignment capability 会进入一次性 executor 环境，但不会进入 label、annotation
+或 OpenSandbox metadata；它只对该 assignment 有效，并且不能绕过 SPIFFE 身份校验。
+
 这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配，
 不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor
 执行已经领取的 task，并继续完成日志、状态、心跳、取消和最终结果上报。
@@ -25,22 +48,57 @@ dynamic-runner scheduler
 ## 设计约束
 
 - 对 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，完成后销毁。
-- SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生，不需要 queued 与
+- SPIFFE 身份从实际领取的 task 的 repository 和 workflow job key 派生，不需要 queued 与
   in-progress webhook 的二阶段关联。
+- 身份中的 task 段使用 workflow job key，而不是可带空格的展示名称；job key 必须满足
+  `[A-Za-z_][A-Za-z0-9_-]*`。slug + hash 只保留为旧名称的显式迁移后备方案。
+- scheduler 只做确定性的身份派生与 executor 绑定，不维护业务授权 policy；Zot、
+  OpenBao 等资源服务继续是唯一授权决策点。
 - scheduler 的 runner registration credential 不进入 executor；executor 只得到执行
   当前 task 所需的短期 lease/capability。
-- task ACK、心跳和结果必须能够跨 scheduler 重启恢复；NATS 可以继续作为内部 handoff，
-  但不是 Gitea 任务事实来源。
+- JetStream 只持久化和投递 assignment，不保存 executor 生命周期状态。Pod labels/annotations
+  与 OpenSandbox metadata 是后端运行状态的权威来源，Gitea 是 task 终态的权威来源。
+- assignment 使用版本化 envelope 保存完整 Gitea protobuf task，并从 workflow `runs-on`
+  严格选择 pod 或 vm subject；消费者解码后重新派生 backend 与身份，拒绝被篡改的冗余字段。
+- JetStream 的 message ID 等于稳定 assignment ID `gitea-task-<task-id>`，仅用于发布去重，
+  不承担 executor 生命周期记录。
+- worker 按稳定 assignment ID reconcile 后端资源，进程内只保留并发控制等可丢弃状态；
+  不新增数据库，也不依赖内存中的 runner-to-executor 映射。
+- VM worker 使用 OpenSandbox 官方 Go SDK，并把 assignment ID、repository、job key 和
+  SPIFFE ID写入 sandbox metadata；通过 `extensions.poolRef=ci-vm` 使用既有 Kata Pool。
+- executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后，facade 在后端
+  metadata 写入持久 terminal marker；backend reconciler 仅在执行环境也进入终态后清理，
+  从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
+- 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 协议，只有环境创建和销毁实现不同。
+- scheduler 在 assignment 持久化到 JetStream 后即可继续领取；Pod 与 VM 分别由 durable
+  consumer 的 capacity 限制并发，不共享全局执行槽位。未知后端故障由对应 consumer 的
+  NAK/redelivery 收敛，不能阻塞另一种 backend。
+- 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生
+  创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。
 
 ## 实现顺序
 
 1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本，记录兼容
    范围并建立协议契约测试。
 2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike，暂不执行
-   task。
+   task。首次集成必须验证 FetchTask 后、JetStream publish 前进程崩溃时 Gitea 对同一
+   runner 的 task 恢复语义；该窗口未验证前不能声称 scheduler 可无损恢复。
 3. 从 act_runner 提取或复用 task 执行与日志上报能力，定义 scheduler 到 executor 的
    单任务协议。
 4. 首先接入 Pod executor，验证成功、失败、取消、超时和 scheduler 重启。

go.mod

@@ -0,0 +1,72 @@
+module git.ddupan.top/panxiao81/gitea-dynamic-runner
+
+go 1.27
+
+require (
+	connectrpc.com/connect v1.20.0
+	gitea.dev/actionslib v1.0.0
+	github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5
+	github.com/nats-io/nats.go v1.54.0
+	github.com/spiffe/go-spiffe/v2 v2.8.2
+	golang.org/x/sync v0.23.0
+	google.golang.org/protobuf v1.36.12
+	k8s.io/api v0.37.0
+	k8s.io/apimachinery v0.37.0
+	k8s.io/client-go v0.37.0
+)
+
+require (
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/emicklei/go-restful/v3 v3.13.0 // indirect
+	github.com/fxamacker/cbor/v2 v2.9.1 // indirect
+	github.com/go-jose/go-jose/v4 v4.1.5 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-openapi/jsonpointer v1.0.0 // indirect
+	github.com/go-openapi/jsonreference v1.0.0 // indirect
+	github.com/go-openapi/swag v0.27.1 // indirect
+	github.com/go-openapi/swag/cmdutils v0.27.1 // indirect
+	github.com/go-openapi/swag/conv v0.27.1 // indirect
+	github.com/go-openapi/swag/fileutils v0.27.1 // indirect
+	github.com/go-openapi/swag/jsonutils v0.27.1 // indirect
+	github.com/go-openapi/swag/loading v0.27.1 // indirect
+	github.com/go-openapi/swag/mangling v0.27.1 // indirect
+	github.com/go-openapi/swag/netutils v0.27.1 // indirect
+	github.com/go-openapi/swag/pools v0.27.1 // indirect
+	github.com/go-openapi/swag/stringutils v0.27.1 // indirect
+	github.com/go-openapi/swag/typeutils v0.27.1 // indirect
+	github.com/go-openapi/swag/yamlutils v0.27.1 // indirect
+	github.com/google/gnostic-models v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.20.0 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
+	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.yaml.in/yaml/v3 v3.0.5 // indirect
+	go.yaml.in/yaml/v4 v4.0.0-rc.6 // indirect
+	golang.org/x/crypto v0.57.0 // indirect
+	golang.org/x/net v0.58.0 // indirect
+	golang.org/x/oauth2 v0.36.0 // indirect
+	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/term v0.46.0 // indirect
+	golang.org/x/text v0.42.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect
+	google.golang.org/grpc v1.79.3 // indirect
+	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	k8s.io/klog/v2 v2.140.0 // indirect
+	k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect
+	k8s.io/utils v0.0.0-20260626114624-be93311217bd // indirect
+	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
+	sigs.k8s.io/randfill v1.0.0 // indirect
+	sigs.k8s.io/structured-merge-diff/v6 v6.4.2 // indirect
+	sigs.k8s.io/yaml v1.6.0 // indirect
+)

go.sum

@@ -0,0 +1,170 @@
+connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
+connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
+gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY=
+gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g=
+github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
+github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
+github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 h1:7mZNkBh4VaI+i4VOTJIknLCBkbGizTkdHYqgcMqeDBM=
+github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5/go.mod h1:w0nIMCTL1L3oSS67ABFOdYxrezYQZhk01UtmKSi2UhA=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes=
+github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
+github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
+github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
+github.com/go-jose/go-jose/v4 v4.1.5 h1:RjgjO2LOtWOJKUC5wpwY9LR3B3vwVAz6JS2YHfYU6eA=
+github.com/go-jose/go-jose/v4 v4.1.5/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s=
+github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y=
+github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY=
+github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0=
+github.com/go-openapi/swag v0.27.1 h1:VotvOLWW8q/EAxB0YdsBBGC8XYyeL1YwBj2ungAGPNg=
+github.com/go-openapi/swag v0.27.1/go.mod h1:GTkJPwHfhJp6MWr4/rCh64HVI3Ofu+tcsbfjfHmTxpE=
+github.com/go-openapi/swag/cmdutils v0.27.1 h1:I7sYqaWVl5mq0NEmNQkAmFDyNin9ufvMX/p2zwtQaOE=
+github.com/go-openapi/swag/cmdutils v0.27.1/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM=
+github.com/go-openapi/swag/conv v0.27.1 h1:8wi9ZG+olmY1wXphl93EWniPtbSPkXM/feH7FgjsvrU=
+github.com/go-openapi/swag/conv v0.27.1/go.mod h1:QbqMivkpKhC3g1B1GGGOJ6ANewI3S62dbzYu3Duowqs=
+github.com/go-openapi/swag/fileutils v0.27.1 h1:QQqBSoi5mW4XpU85nS0mLcA+zAE6vLzrb0QkmLKf9oM=
+github.com/go-openapi/swag/fileutils v0.27.1/go.mod h1:VvJFZLTZS0AI854gEQz5tk7dBESdLjiNUMSZ/th2ry8=
+github.com/go-openapi/swag/jsonutils v0.27.1 h1:SVgK3i4USzCU5mibOOS/l4ea2h9UQXy7J7RNLTjuXjU=
+github.com/go-openapi/swag/jsonutils v0.27.1/go.mod h1:tdlEpZqdcQ17uj6J4YdK9vd8It5qWMwjWXOs0tjpRlk=
+github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1 h1:mJu3COL9WEaZVp/Kf2PRMi7tPszPEJfSr/OO75ynCs8=
+github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1/go.mod h1:mofwUWx70wvskwESqRJ//k/9kURmCgyJl5m5Ppoh5kY=
+github.com/go-openapi/swag/loading v0.27.1 h1:/DxUgDXKbBX4bcn7r9uEXfJyzN5XpiJmZplzQTjrRCY=
+github.com/go-openapi/swag/loading v0.27.1/go.mod h1:jvGh3iA2+zyUUycB5fgJWzeHnhrpvGnJJM0RVE9ZShE=
+github.com/go-openapi/swag/mangling v0.27.1 h1:yC9D0HyUE8gbP+BfmGx9+AA89ikwZTMjESK3OnnoaqA=
+github.com/go-openapi/swag/mangling v0.27.1/go.mod h1:jtBE2+V+3pILxOR7Vgce+Cwp6A2PgZbvVqfNntbVs0w=
+github.com/go-openapi/swag/netutils v0.27.1 h1:mICMFoS82F5TZ4Zy3cqmcQk+BFeCp3Uyq3Np7GI0/qU=
+github.com/go-openapi/swag/netutils v0.27.1/go.mod h1:J+WYyFMLtvtCGqa6jLv+YNUmIKI3ZRQRrvfNDMoQoEQ=
+github.com/go-openapi/swag/pools v0.27.1 h1:9LeadcMyb2GJCbXX5hVQDbZ2Lq9TL4dCs/nx1j5DO0E=
+github.com/go-openapi/swag/pools v0.27.1/go.mod h1:kVQefhSK5RWuRe7BXsL8htgBPAMpN7HDGpGEknqugeE=
+github.com/go-openapi/swag/stringutils v0.27.1 h1:ZXePZ0r2p1qSjo8tD3Un4vFj8+FqlCkczxDrJIhYUp8=
+github.com/go-openapi/swag/stringutils v0.27.1/go.mod h1:lzRN95CxXmA03XcDWHLOb6nOMcxCqR5rGY0lOgsfRoM=
+github.com/go-openapi/swag/typeutils v0.27.1 h1:KSTdFlfnse4r6dP9IrEnwMldjE+zs71UeEB3//PtVXc=
+github.com/go-openapi/swag/typeutils v0.27.1/go.mod h1:Srm0xFNRZ1Y+vCxJclo5qzx8aj+1pAKda/YfFPrG0dQ=
+github.com/go-openapi/swag/yamlutils v0.27.1 h1:ftxv6xvXb1E3zohUc+okZ9nSqNb9StQX/FXnKZ98sQA=
+github.com/go-openapi/swag/yamlutils v0.27.1/go.mod h1:bnxFIB1qewGRiZHypXGZ3fNgf13/0HfRgnS/iZBDrOo=
+github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL68dAVi84WuOApp29B0=
+github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo=
+github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug=
+github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
+github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8=
+github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
+github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
+github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.10.2 h1:G2SED73/qrAu6YwbdxOD6peLkCBI3z7L+ykJFTXJBBo=
+github.com/sirupsen/logrus v1.10.2/go.mod h1:SLEg8TqYulVKKfIGHldVp2K2aYz2DKSVBq4g/H5bR7Q=
+github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
+github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/spiffe/go-spiffe/v2 v2.8.2 h1:jUEsvCMD6fH25J8K/w3q/XnIx8W1lb8+YLaEEHIjHmc=
+github.com/spiffe/go-spiffe/v2 v2.8.2/go.mod h1:w2CLWKLMTX/PPYUEUPv3ltH0RXsw5S8suwNF46w9/Aw=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
+github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
+github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
+github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
+github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
+go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
+go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
+go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
+go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
+go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE=
+go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8=
+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=
+go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
+go.yaml.in/yaml/v4 v4.0.0-rc.6 h1:1h7H1ohdUh93/FyE4YaDa1Zh64K6VVbjF4K6WUxMtH4=
+go.yaml.in/yaml/v4 v4.0.0-rc.6/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0=
+golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
+golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
+golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
+golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
+golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk=
+golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
+golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
+golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
+golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
+gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
+google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
+google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
+google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo=
+gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M=
+gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
+gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
+k8s.io/api v0.37.0 h1:Z//Vj9N7RA/yS2sDmxyeo7h+RR4zbUrd2vrd3Z0TbB4=
+k8s.io/api v0.37.0/go.mod h1:LKXgcJWMc+f4OLbP5SFR8rulEg07zZhpi/zMULiBImk=
+k8s.io/apimachinery v0.37.0 h1:Np2AbDtf8x6RDHiD8T9LbKJ9gaegeVNa8yNm5FuGKm0=
+k8s.io/apimachinery v0.37.0/go.mod h1:RN3nhprFSCxOi5Selxd7oMTXOe/c+ZbcE7Im+TS2zkE=
+k8s.io/client-go v0.37.0 h1:nsN31fy8wBySuZ+QRnKmrjRSQLOG2rvoGN0tKd12zhQ=
+k8s.io/client-go v0.37.0/go.mod h1:FcGqw+Ll/gNQiq+nPGY1Oyt9y7SgDh1d3MW3RFDEbn0=
+k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc=
+k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0=
+k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad h1:oXImqH8mQNk7PmvzKhmN3ddJoY6OnyM225MXwGHPm0A=
+k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad/go.mod h1:0/mqHCVhlumdJ3BhCfnjSZQE037nAhNodh1/hK0T8/I=
+k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI07D/dEYRaB9ZZEs=
+k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk=
+sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
+sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
+sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
+sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
+sigs.k8s.io/structured-merge-diff/v6 v6.4.2 h1:qdOxHwrl2Kaag1aQEarlYcOA9vSyGCp3CIki3aW8c4Q=
+sigs.k8s.io/structured-merge-diff/v6 v6.4.2/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
+sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs=
+sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4=

internal/assignmentqueue/jetstream.go

@@ -0,0 +1,238 @@
+// Package assignmentqueue implements the durable assignment handoff with JetStream.
+package assignmentqueue
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type publishAPI interface {
+	PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error)
+}
+
+// Publisher implements the scheduler dispatcher with one subject per backend.
+type Publisher struct {
+	JetStream   publishAPI
+	SubjectBase string
+}
+
+func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assignment) error {
+	if p.JetStream == nil {
+		return errors.New("JetStream publisher is required")
+	}
+	body, err := taskassignment.Marshal(assignment)
+	if err != nil {
+		return err
+	}
+	base := strings.TrimSuffix(p.SubjectBase, ".")
+	if base == "" {
+		return errors.New("assignment subject base is required")
+	}
+	message := &nats.Msg{
+		Subject: base + "." + string(assignment.Backend),
+		Header:  nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}},
+		Data:    body,
+	}
+	if _, err := p.JetStream.PublishMsg(ctx, message); err != nil {
+		return fmt.Errorf("publish assignment %s: %w", assignment.ID, err)
+	}
+	return nil
+}
+
+type Accepter interface {
+	Accept(context.Context, taskassignment.Assignment) (bool, error)
+}
+
+type Claims interface {
+	Offer(taskassignment.Assignment) (<-chan struct{}, error)
+	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
+	DoubleAck(context.Context) error
+	NakWithDelay(time.Duration) error
+	TermWithReason(string) error
+}
+
+// Processor maps one delivery to one idempotent worker reconciliation.
+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 || 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
+		}
+		claimContext, cancel := context.WithTimeout(ctx, timeout)
+		err := p.Claims.WaitClaimed(claimContext, assignment.ID)
+		cancel()
+		if err != nil {
+			delay := p.RetryDelay
+			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)
+}
+
+type consumeAPI interface {
+	Consume(jetstream.MessageHandler, ...jetstream.PullConsumeOpt) (jetstream.ConsumeContext, error)
+}
+
+// ConsumerComponent runs bounded reconciliation goroutines for one durable
+// backend consumer. The goroutine set is operational state, not task storage.
+type ConsumerComponent struct {
+	Consumer  consumeAPI
+	Processor Processor
+	Capacity  int
+	OnError   func(error)
+}
+
+type consumerManager interface {
+	CreateOrUpdateConsumer(context.Context, string, jetstream.ConsumerConfig) (jetstream.Consumer, error)
+}
+
+// OpenConsumer creates the durable backend cursor. Capacity is enforced both
+// server-side and by ConsumerComponent's local semaphore.
+func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, backend taskassignment.Backend, capacity int) (jetstream.Consumer, error) {
+	if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 {
+		return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required")
+	}
+	if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
+		return nil, fmt.Errorf("unsupported assignment backend %q", backend)
+	}
+	consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
+		Name:          string(backend),
+		Durable:       string(backend),
+		FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + string(backend),
+		AckPolicy:     jetstream.AckExplicitPolicy,
+		AckWait:       5 * time.Minute,
+		MaxAckPending: capacity,
+		MaxDeliver:    1000,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
+	}
+	return consumer, nil
+}
+
+func (c ConsumerComponent) Run(ctx context.Context) error {
+	if c.Consumer == nil || c.Capacity < 1 {
+		return errors.New("JetStream consumer and positive capacity are required")
+	}
+	semaphore := make(chan struct{}, c.Capacity)
+	var workers sync.WaitGroup
+	consumeContext, err := c.Consumer.Consume(func(message jetstream.Msg) {
+		select {
+		case semaphore <- struct{}{}:
+		case <-ctx.Done():
+			return
+		}
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			defer func() { <-semaphore }()
+			if err := c.Processor.Process(ctx, message); err != nil && !errors.Is(err, context.Canceled) && c.OnError != nil {
+				c.OnError(err)
+			}
+		}()
+	}, jetstream.PullMaxMessages(c.Capacity))
+	if err != nil {
+		return fmt.Errorf("start JetStream consumer: %w", err)
+	}
+
+	select {
+	case <-ctx.Done():
+		consumeContext.Stop()
+		<-consumeContext.Closed()
+		workers.Wait()
+		return nil
+	case <-consumeContext.Closed():
+		workers.Wait()
+		return errors.New("JetStream consumer stopped unexpectedly")
+	}
+}

internal/assignmentqueue/jetstream_test.go

@@ -0,0 +1,221 @@
+package assignmentqueue
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+func testAssignment(t *testing.T) taskassignment.Assignment {
+	t.Helper()
+	fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assignment, err := taskassignment.New(&runnerv1.Task{
+		Id:              42,
+		Context:         fields,
+		WorkflowPayload: []byte("jobs:\n  publish:\n    runs-on: [self-hosted, pod]\n    steps: []\n"),
+	}, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return assignment
+}
+
+type fakePublisher struct{ message *nats.Msg }
+
+func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
+	p.message = message
+	return &jetstream.PubAck{}, nil
+}
+
+func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) {
+	api := &fakePublisher{}
+	publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"}
+	if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil {
+		t.Fatal(err)
+	}
+	if api.message.Subject != "ci.assignment.pod" {
+		t.Fatalf("subject = %q", api.message.Subject)
+	}
+	if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" {
+		t.Fatalf("message ID = %q", api.message.Header.Get(jetstream.MsgIDHeader))
+	}
+}
+
+type fakeAccepter struct {
+	accepted bool
+	err      error
+}
+
+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 {
+		close(ready)
+	}
+	return ready, nil
+}
+
+func (c *fakeClaims) WaitClaimed(ctx context.Context, _ string) error {
+	if c.claimed {
+		return nil
+	}
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func (a *fakeAccepter) Accept(context.Context, taskassignment.Assignment) (bool, error) {
+	return a.accepted, a.err
+}
+
+type fakeMessage struct {
+	data       []byte
+	acked      int
+	nacked     time.Duration
+	terminated int
+}
+
+func (m *fakeMessage) Data() []byte                           { return m.data }
+func (m *fakeMessage) DoubleAck(context.Context) error        { m.acked++; return nil }
+func (m *fakeMessage) NakWithDelay(delay time.Duration) error { m.nacked = delay; return nil }
+func (m *fakeMessage) TermWithReason(string) error            { m.terminated++; return nil }
+
+func encodedAssignment(t *testing.T) []byte {
+	t.Helper()
+	data, err := taskassignment.Marshal(testAssignment(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return data
+}
+
+func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
+	message := &fakeMessage{data: encodedAssignment(t)}
+	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{}, Admission: &fakeAdmission{allowed: true}, RetryDelay: 2 * time.Second}
+	if err := processor.Process(context.Background(), message); err != nil {
+		t.Fatal(err)
+	}
+	if message.acked != 0 || message.nacked != 2*time.Second {
+		t.Fatalf("message = %#v", message)
+	}
+}
+
+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 {
+		t.Fatal("expected backend error")
+	}
+	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 {
+		t.Fatal("expected decode error")
+	}
+	if poison.terminated != 1 || poison.nacked != 0 {
+		t.Fatalf("poison message = %#v", poison)
+	}
+}
+
+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) {
+	m.config = config
+	return nil, nil
+}
+
+func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
+	manager := &fakeConsumerManager{}
+	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 || 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/controller/components.go

@@ -0,0 +1,75 @@
+// Package controller composes independently runnable scheduler and backend workers.
+package controller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+
+	"golang.org/x/sync/errgroup"
+)
+
+type ComponentName string
+
+const (
+	Scheduler ComponentName = "scheduler"
+	PodWorker ComponentName = "pod-worker"
+	VMWorker  ComponentName = "vm-worker"
+)
+
+var defaultComponents = []ComponentName{Scheduler, PodWorker, VMWorker}
+
+// Selection parses --components. An empty value enables all components.
+type Selection []ComponentName
+
+func ParseSelection(value string) (Selection, error) {
+	if strings.TrimSpace(value) == "" || strings.TrimSpace(value) == "all" {
+		return append(Selection(nil), defaultComponents...), nil
+	}
+	var selected Selection
+	for _, raw := range strings.Split(value, ",") {
+		name := ComponentName(strings.TrimSpace(raw))
+		if !slices.Contains(defaultComponents, name) {
+			return nil, fmt.Errorf("unknown controller component %q", name)
+		}
+		if !slices.Contains(selected, name) {
+			selected = append(selected, name)
+		}
+	}
+	if len(selected) == 0 {
+		return nil, errors.New("at least one controller component is required")
+	}
+	return selected, nil
+}
+
+type Component interface {
+	Run(context.Context) error
+}
+
+type Registry map[ComponentName]Component
+
+// Run starts exactly the selected components in one process. The first real
+// failure cancels its peers; ordinary context cancellation is graceful.
+func Run(ctx context.Context, selection Selection, registry Registry) error {
+	group, groupContext := errgroup.WithContext(ctx)
+	for _, name := range selection {
+		component, ok := registry[name]
+		if !ok || component == nil {
+			return fmt.Errorf("component %q is not configured", name)
+		}
+		name, component := name, component
+		group.Go(func() error {
+			err := component.Run(groupContext)
+			if errors.Is(err, context.Canceled) && groupContext.Err() != nil {
+				return nil
+			}
+			if err != nil {
+				return fmt.Errorf("component %s: %w", name, err)
+			}
+			return nil
+		})
+	}
+	return group.Wait()
+}

internal/controller/components_test.go

@@ -0,0 +1,63 @@
+package controller
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+)
+
+func TestParseSelectionDefaultsToAll(t *testing.T) {
+	for _, input := range []string{"", "all"} {
+		selection, err := ParseSelection(input)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(selection) != 3 || selection[0] != Scheduler || selection[1] != PodWorker || selection[2] != VMWorker {
+			t.Fatalf("selection = %v", selection)
+		}
+	}
+}
+
+func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) {
+	selection, err := ParseSelection("vm-worker,scheduler,vm-worker")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(selection) != 2 || selection[0] != VMWorker || selection[1] != Scheduler {
+		t.Fatalf("selection = %v", selection)
+	}
+	if _, err := ParseSelection("webhook"); err == nil {
+		t.Fatal("expected obsolete component to be rejected")
+	}
+}
+
+type componentFunc func(context.Context) error
+
+func (f componentFunc) Run(ctx context.Context) error { return f(ctx) }
+
+func TestRunStartsSelectedComponentsAndCancelsPeers(t *testing.T) {
+	started := make(chan ComponentName, 2)
+	peerStopped := make(chan struct{})
+	var once sync.Once
+	registry := Registry{
+		Scheduler: componentFunc(func(context.Context) error {
+			started <- Scheduler
+			return errors.New("poll failed")
+		}),
+		PodWorker: componentFunc(func(ctx context.Context) error {
+			started <- PodWorker
+			<-ctx.Done()
+			once.Do(func() { close(peerStopped) })
+			return ctx.Err()
+		}),
+	}
+	err := Run(context.Background(), Selection{Scheduler, PodWorker}, registry)
+	if err == nil || !errors.Is(err, context.Canceled) && err.Error() != "component scheduler: poll failed" {
+		t.Fatalf("Run() error = %v", err)
+	}
+	<-peerStopped
+	if len(started) != 2 {
+		t.Fatalf("started components = %d", len(started))
+	}
+}

internal/giteaactions/client.go

@@ -0,0 +1,71 @@
+// Package giteaactions provides the authenticated Gitea RunnerService client.
+package giteaactions
+
+import (
+	"context"
+	"net/http"
+	"strings"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+)
+
+// Client is the subset of RunnerService owned by the scheduler.
+type Client struct {
+	runner runnerv1connect.RunnerServiceClient
+}
+
+// NewClient authenticates every RPC with the persistent scheduler runner.
+func NewClient(httpClient connect.HTTPClient, instanceURL, uuid, token string) *Client {
+	auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
+		return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+			request.Header().Set("User-Agent", "gitea-dynamic-runner-go/0")
+			request.Header().Set(protocol.UUIDHeader, uuid)
+			request.Header().Set(protocol.TokenHeader, token)
+			return next(ctx, request)
+		}
+	})
+	baseURL := strings.TrimRight(instanceURL, "/") + "/api/actions"
+	return &Client{runner: runnerv1connect.NewRunnerServiceClient(
+		httpClient,
+		baseURL,
+		connect.WithInterceptors(auth),
+	)}
+}
+
+// Declare advertises the scheduler labels before tasks are fetched.
+func (c *Client) Declare(ctx context.Context, version string, labels []string) error {
+	_, err := c.runner.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
+		Version: version,
+		Labels:  labels,
+	}))
+	return err
+}
+
+// FetchTask asks Gitea to atomically assign the next matching task.
+func (c *Client) FetchTask(ctx context.Context, tasksVersion int64) (*runnerv1.FetchTaskResponse, error) {
+	response, err := c.runner.FetchTask(ctx, connect.NewRequest(&runnerv1.FetchTaskRequest{
+		TasksVersion: tasksVersion,
+	}))
+	if err != nil {
+		return nil, err
+	}
+	return response.Msg, nil
+}
+
+// UpdateTask forwards executor state through the scheduler runner identity.
+func (c *Client) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
+	return c.runner.UpdateTask(ctx, request)
+}
+
+// UpdateLog forwards executor log rows through the scheduler runner identity.
+func (c *Client) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
+	return c.runner.UpdateLog(ctx, request)
+}
+
+// DefaultHTTPClient is suitable for the scheduler's long-lived connection.
+func DefaultHTTPClient() *http.Client {
+	return &http.Client{Transport: http.DefaultTransport}
+}

internal/giteaactions/client_test.go

@@ -0,0 +1,101 @@
+package giteaactions
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+)
+
+type runnerService struct {
+	runnerv1connect.UnimplementedRunnerServiceHandler
+	t              *testing.T
+	declaredLabels []string
+	fetchedVersion int64
+	updatedTask    int64
+	updatedLog     int64
+	expectedUUID   string
+	expectedToken  string
+}
+
+func (s *runnerService) checkAuth(request connect.AnyRequest) {
+	s.t.Helper()
+	if got := request.Header().Get(protocol.UUIDHeader); got != s.expectedUUID {
+		s.t.Errorf("runner UUID header = %q", got)
+	}
+	if got := request.Header().Get(protocol.TokenHeader); got != s.expectedToken {
+		s.t.Errorf("runner token header = %q", got)
+	}
+}
+
+func (s *runnerService) Declare(_ context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
+	s.checkAuth(request)
+	s.declaredLabels = request.Msg.Labels
+	return connect.NewResponse(&runnerv1.DeclareResponse{}), nil
+}
+
+func (s *runnerService) FetchTask(_ context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
+	s.checkAuth(request)
+	s.fetchedVersion = request.Msg.TasksVersion
+	return connect.NewResponse(&runnerv1.FetchTaskResponse{
+		Task:         &runnerv1.Task{Id: 42},
+		TasksVersion: 8,
+	}), nil
+}
+
+func (s *runnerService) UpdateTask(_ context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
+	s.checkAuth(request)
+	s.updatedTask = request.Msg.GetState().GetId()
+	return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: request.Msg.State}), nil
+}
+
+func (s *runnerService) UpdateLog(_ context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
+	s.checkAuth(request)
+	s.updatedLog = request.Msg.GetTaskId()
+	return connect.NewResponse(&runnerv1.UpdateLogResponse{}), nil
+}
+
+func TestClientUsesOfficialRunnerProtocol(t *testing.T) {
+	service := &runnerService{
+		t:             t,
+		expectedUUID:  "runner-uuid",
+		expectedToken: "runner-token",
+	}
+	path, handler := runnerv1connect.NewRunnerServiceHandler(service)
+	mux := http.NewServeMux()
+	mux.Handle("/api/actions"+path, http.StripPrefix("/api/actions", handler))
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	client := NewClient(server.Client(), server.URL, service.expectedUUID, service.expectedToken)
+	labels := []string{"self-hosted", "pod", "vm"}
+	if err := client.Declare(context.Background(), "0.1.0", labels); err != nil {
+		t.Fatal(err)
+	}
+	response, err := client.FetchTask(context.Background(), 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(service.declaredLabels, labels) {
+		t.Fatalf("declared labels = %#v", service.declaredLabels)
+	}
+	if service.fetchedVersion != 7 || response.GetTasksVersion() != 8 || response.GetTask().GetId() != 42 {
+		t.Fatalf("unexpected FetchTask exchange: request=%d response=%v", service.fetchedVersion, response)
+	}
+	if _, err := client.UpdateTask(context.Background(), connect.NewRequest(&runnerv1.UpdateTaskRequest{State: &runnerv1.TaskState{Id: 42}})); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := client.UpdateLog(context.Background(), connect.NewRequest(&runnerv1.UpdateLogRequest{TaskId: 42})); err != nil {
+		t.Fatal(err)
+	}
+	if service.updatedTask != 42 || service.updatedLog != 42 {
+		t.Fatalf("updated task=%d log=%d", service.updatedTask, service.updatedLog)
+	}
+}

internal/opensandboxbackend/backend.go

@@ -0,0 +1,310 @@
+// Package opensandboxbackend implements the VM executor backend with the official OpenSandbox SDK.
+package opensandboxbackend
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+	"time"
+
+	opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
+)
+
+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)
+	CreateSandbox(context.Context, opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error)
+	GetSandbox(context.Context, string) (*opensandbox.SandboxInfo, error)
+	PatchSandboxMetadata(context.Context, string, opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error)
+	DeleteSandbox(context.Context, string) error
+}
+
+// MarkTerminal persists the accepted Gitea terminal state on the sandbox. The
+// lifecycle reconciler performs deletion separately so the runner receives the
+// successful UpdateTask response before its VM is stopped.
+func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error {
+	executor, err := b.Find(ctx, assignmentID)
+	if err != nil || executor == nil {
+		return err
+	}
+	value := "true"
+	_, err = b.Lifecycle.PatchSandboxMetadata(ctx, executor.Name, opensandbox.MetadataPatch{
+		terminalMetadata: &value,
+	})
+	if err != nil {
+		return fmt.Errorf("mark sandbox %s terminal: %w", executor.Name, err)
+	}
+	return nil
+}
+
+// CleanupTerminated removes sandboxes whose terminal result was accepted by
+// Gitea. The marker is stored by OpenSandbox, so cleanup survives restarts.
+func (b Backend) CleanupTerminated(ctx context.Context) (int, error) {
+	if err := b.validate(); err != nil {
+		return 0, err
+	}
+	result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
+		Metadata: map[string]string{terminalMetadata: "true"},
+		PageSize: 100,
+	})
+	if err != nil {
+		return 0, fmt.Errorf("list terminal sandboxes: %w", err)
+	}
+	cleaned := 0
+	for _, sandbox := range result.Items {
+		if sandbox.Metadata[assignmentMetadata] == "" {
+			continue
+		}
+		if err := b.Delete(ctx, executor(sandbox)); err != nil {
+			return cleaned, fmt.Errorf("delete terminal sandbox %s: %w", sandbox.ID, err)
+		}
+		cleaned++
+	}
+	return cleaned, nil
+}
+
+// Lifecycle periodically reconciles durable terminal markers into deletes.
+type LifecycleReconciler struct {
+	Backend  Backend
+	Interval time.Duration
+	OnError  func(error)
+}
+
+func (l LifecycleReconciler) Run(ctx context.Context) error {
+	interval := l.Interval
+	if interval <= 0 {
+		interval = 2 * time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil {
+			l.OnError(err)
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+		}
+	}
+}
+
+type Config struct {
+	Pool       string
+	Timeout    int
+	Entrypoint []string
+	Env        map[string]string
+}
+
+type Backend struct {
+	Lifecycle Lifecycle
+	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))
+	}
+	return opensandbox.NewLifecycleClient(baseURL, apiKey)
+}
+
+func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
+		Metadata: map[string]string{assignmentMetadata: assignmentID}, PageSize: 2,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list assignment sandboxes: %w", err)
+	}
+	if len(result.Items) > 1 {
+		return nil, fmt.Errorf("assignment %s owns %d sandboxes", assignmentID, len(result.Items))
+	}
+	if len(result.Items) == 0 {
+		return nil, nil
+	}
+	return executor(result.Items[0]), nil
+}
+
+func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	if assignment.Backend != taskassignment.BackendVM {
+		return nil, fmt.Errorf("OpenSandbox backend cannot create %q assignment", assignment.Backend)
+	}
+	environment := clone(b.Config.Env)
+	for key, value := range launch.Environment {
+		environment[key] = value
+	}
+	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{
+		Timeout: &b.Config.Timeout, Entrypoint: append([]string{}, b.Config.Entrypoint...),
+		Env: environment, Metadata: sandboxMetadata,
+		Extensions:     map[string]string{"poolRef": b.Config.Pool},
+		ResourceLimits: opensandbox.ResourceLimits{},
+	}
+	sandbox, err := b.Lifecycle.CreateSandbox(ctx, request)
+	if err != nil {
+		return nil, fmt.Errorf("create assignment sandbox: %w", err)
+	}
+	return executor(*sandbox), nil
+}
+
+func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error {
+	if executor == nil || executor.Name == "" {
+		return errors.New("sandbox ID is required for identity binding")
+	}
+	sandbox, err := b.Lifecycle.GetSandbox(ctx, executor.Name)
+	if err != nil {
+		return fmt.Errorf("verify sandbox identity metadata: %w", err)
+	}
+	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
+}
+
+func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error {
+	if executor == nil || executor.Name == "" {
+		return nil
+	}
+	err := b.Lifecycle.DeleteSandbox(ctx, executor.Name)
+	var apiError *opensandbox.APIError
+	if errors.As(err, &apiError) && apiError.StatusCode == http.StatusNotFound {
+		return nil
+	}
+	return err
+}
+
+func (b Backend) validate() error {
+	if b.Lifecycle == nil || b.Config.Pool == "" || b.Config.Timeout < 1 || len(b.Config.Entrypoint) == 0 {
+		return errors.New("OpenSandbox Lifecycle client, pool, timeout, and entrypoint are required")
+	}
+	return nil
+}
+
+func executor(sandbox opensandbox.SandboxInfo) *taskworker.Executor {
+	return &taskworker.Executor{Name: sandbox.ID, IdentityTarget: sandbox.ID, Phase: phase(sandbox.Status.State)}
+}
+
+func phase(state opensandbox.SandboxState) taskworker.Phase {
+	switch state {
+	case opensandbox.StateRunning:
+		return taskworker.PhaseRunning
+	case opensandbox.StateTerminated, opensandbox.StateFailed:
+		// A successful executor reports Gitea before it exits. If Gitea is not
+		// terminal when the sandbox stops, termination is an execution failure.
+		return taskworker.PhaseFailed
+	default:
+		return taskworker.PhasePending
+	}
+}
+
+func clone(source map[string]string) map[string]string {
+	result := make(map[string]string, len(source))
+	for key, value := range source {
+		result[key] = value
+	}
+	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

@@ -0,0 +1,150 @@
+package opensandboxbackend
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
+)
+
+type fakeLifecycle struct {
+	items   []opensandbox.SandboxInfo
+	created opensandbox.CreateSandboxRequest
+	deleted string
+}
+
+func (f *fakeLifecycle) ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error) {
+	return &opensandbox.ListSandboxesResponse{Items: f.items}, nil
+}
+func (f *fakeLifecycle) CreateSandbox(_ context.Context, request opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error) {
+	f.created = request
+	return &opensandbox.SandboxInfo{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StatePending}, Metadata: request.Metadata}, nil
+}
+func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.SandboxInfo, error) {
+	for _, item := range f.items {
+		if item.ID == id {
+			return &item, 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 {
+		if f.items[index].ID != id {
+			continue
+		}
+		if f.items[index].Metadata == nil {
+			f.items[index].Metadata = map[string]string{}
+		}
+		for key, value := range patch {
+			if value != nil {
+				f.items[index].Metadata[key] = *value
+			}
+		}
+		return &f.items[index], nil
+	}
+	return &opensandbox.SandboxInfo{ID: id}, nil
+}
+func (f *fakeLifecycle) DeleteSandbox(_ context.Context, id string) error { f.deleted = id; return nil }
+
+func backend(lifecycle Lifecycle) Backend {
+	return Backend{Lifecycle: lifecycle, Config: Config{
+		Pool: "ci-vm", Timeout: 14400,
+		Entrypoint: []string{"/usr/local/bin/gitea-task-executor"},
+		Env:        map[string]string{"SPIFFE_ENDPOINT_SOCKET": "unix:///run/spire/agent-sockets/spire-agent.sock"},
+	}}
+}
+
+func assignment() taskassignment.Assignment {
+	return taskassignment.Assignment{
+		ID: "gitea-task-42", Backend: taskassignment.BackendVM,
+		Task:     &runnerv1.Task{Id: 42},
+		Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
+	}
+}
+
+func TestFindRecoversSandboxByMetadata(t *testing.T) {
+	lifecycle := &fakeLifecycle{items: []opensandbox.SandboxInfo{{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StateRunning}}}}
+	executor, err := backend(lifecycle).Find(context.Background(), assignment().ID)
+	if err != nil || executor.Name != "sandbox-42" || executor.Phase != taskworker.PhaseRunning {
+		t.Fatalf("executor=%#v err=%v", executor, err)
+	}
+}
+
+func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) {
+	lifecycle := &fakeLifecycle{}
+	metadata := taskworker.BackendMetadata(assignment())
+	executor, err := backend(lifecycle).Create(context.Background(), assignment(), taskworker.LaunchSpec{
+		Metadata: metadata,
+		Environment: map[string]string{
+			"CI_SPIFFE_ID":         assignment().Identity.SPIFFEID,
+			"CI_RUNNER_CAPABILITY": "capability",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	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) {
+	lifecycle := &fakeLifecycle{}
+	if err := backend(lifecycle).BindIdentity(context.Background(), &taskworker.Executor{Name: "sandbox-42"}, assignment().Identity); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestTerminalMarkerDrivesDurableCleanup(t *testing.T) {
+	lifecycle := &fakeLifecycle{items: []opensandbox.SandboxInfo{{
+		ID:       "sandbox-42",
+		Metadata: map[string]string{assignmentMetadata: assignment().ID},
+	}}}
+	backend := backend(lifecycle)
+	if err := backend.MarkTerminal(context.Background(), assignment().ID); err != nil {
+		t.Fatal(err)
+	}
+	if lifecycle.items[0].Metadata[terminalMetadata] != "true" {
+		t.Fatalf("metadata = %#v", lifecycle.items[0].Metadata)
+	}
+	cleaned, err := backend.CleanupTerminated(context.Background())
+	if err != nil || cleaned != 1 || lifecycle.deleted != "sandbox-42" {
+		t.Fatalf("cleaned=%d deleted=%q err=%v", cleaned, lifecycle.deleted, err)
+	}
+}

internal/podbackend/backend.go

@@ -0,0 +1,266 @@
+// Package podbackend implements the native homelab Kubernetes executor backend.
+package podbackend
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
+)
+
+const (
+	assignmentLabel = "ci.ddupan.top/assignment-id"
+	terminalLabel   = "ci.ddupan.top/terminal"
+)
+
+// Pod is the backend state required by the reconciler, not an in-memory lifecycle record.
+type Pod struct {
+	Name        string
+	Namespace   string
+	UID         string
+	Phase       string
+	Labels      map[string]string
+	Annotations map[string]string
+}
+
+// PodManifest leaves task execution wiring to the executor image while fixing
+// the metadata contract needed for recovery.
+type PodManifest struct {
+	Name           string
+	Namespace      string
+	Labels         map[string]string
+	Annotations    map[string]string
+	Image          string
+	ServiceAccount string
+	Args           []string
+	Environment    map[string]string
+}
+
+// IdentityEntry is a ClusterStaticEntry pinned to one concrete Pod UID.
+type IdentityEntry struct {
+	Name      string
+	Labels    map[string]string
+	ClassName string
+	ParentID  string
+	SPIFFEID  string
+	Selectors []string
+}
+
+// API is the narrow Kubernetes boundary used by the backend adapter.
+type API interface {
+	ListPods(context.Context, string, string) ([]Pod, error)
+	CreatePod(context.Context, PodManifest) (Pod, error)
+	DeletePod(context.Context, string, string) error
+	LabelPod(context.Context, string, string, map[string]string) error
+	EnsureIdentityEntry(context.Context, IdentityEntry) error
+	DeleteIdentityEntry(context.Context, string) error
+}
+
+// MarkTerminal persists Gitea's accepted terminal state on the backend
+// resource. Cleanup can therefore resume after a controller restart.
+func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error {
+	if err := b.validate(); err != nil {
+		return err
+	}
+	if assignmentID == "" {
+		return errors.New("assignment ID is required")
+	}
+	if err := b.API.LabelPod(ctx, b.Config.Namespace, assignmentID, map[string]string{terminalLabel: "true"}); err != nil {
+		return fmt.Errorf("mark assignment Pod terminal: %w", err)
+	}
+	return nil
+}
+
+// CleanupTerminated removes only executors whose terminal update was accepted
+// by Gitea and whose process has exited.
+func (b Backend) CleanupTerminated(ctx context.Context) (int, error) {
+	if err := b.validate(); err != nil {
+		return 0, err
+	}
+	pods, err := b.API.ListPods(ctx, b.Config.Namespace, terminalLabel+"=true")
+	if err != nil {
+		return 0, fmt.Errorf("list terminal assignment Pods: %w", err)
+	}
+	cleaned := 0
+	for _, pod := range pods {
+		if pod.Phase != "Succeeded" && pod.Phase != "Failed" {
+			continue
+		}
+		if err := b.Delete(ctx, executor(pod)); err != nil {
+			return cleaned, err
+		}
+		cleaned++
+	}
+	return cleaned, nil
+}
+
+type Config struct {
+	Namespace      string
+	Image          string
+	ServiceAccount string
+	ExecutorArgs   []string
+	TrustDomain    string
+	SPIRECluster   string
+	SPIREClass     string
+	SPIREAgentID   string
+	ExecutorUID    int
+}
+
+type Backend struct {
+	API    API
+	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
+	}
+	pods, err := b.API.ListPods(ctx, b.Config.Namespace, assignmentLabel+"="+assignmentID)
+	if err != nil {
+		return nil, fmt.Errorf("list assignment Pods: %w", err)
+	}
+	if len(pods) > 1 {
+		return nil, fmt.Errorf("assignment %s owns %d Pods", assignmentID, len(pods))
+	}
+	if len(pods) == 0 {
+		return nil, nil
+	}
+	return executor(pods[0]), nil
+}
+
+func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	if assignment.Backend != taskassignment.BackendPod {
+		return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend)
+	}
+	labels := clone(launch.Metadata.Labels)
+	labels["app.kubernetes.io/name"] = "gitea-dynamic-runner"
+	labels["app.kubernetes.io/component"] = "executor"
+	pod, err := b.API.CreatePod(ctx, PodManifest{
+		Name:           assignment.ID,
+		Namespace:      b.Config.Namespace,
+		Labels:         labels,
+		Annotations:    clone(launch.Metadata.Annotations),
+		Image:          b.Config.Image,
+		ServiceAccount: b.Config.ServiceAccount,
+		Args:           append([]string{}, b.Config.ExecutorArgs...),
+		Environment:    clone(launch.Environment),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create assignment Pod: %w", err)
+	}
+	return executor(pod), nil
+}
+
+func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error {
+	if err := b.validate(); err != nil {
+		return err
+	}
+	if executor == nil || executor.Name == "" || executor.IdentityTarget == "" {
+		return errors.New("Pod name and UID are required for identity binding")
+	}
+	if _, err := identityPath(identity.SPIFFEID, b.Config.TrustDomain); err != nil {
+		return err
+	}
+	return b.API.EnsureIdentityEntry(ctx, IdentityEntry{
+		Name: executor.Name,
+		Labels: map[string]string{
+			"app.kubernetes.io/name":      "gitea-dynamic-runner",
+			"app.kubernetes.io/component": "pod-identity",
+			assignmentLabel:               executor.Name,
+		},
+		ClassName: b.Config.SPIREClass,
+		ParentID:  b.Config.SPIREAgentID,
+		SPIFFEID:  identity.SPIFFEID,
+		Selectors: []string{"k8s:pod-uid:" + executor.IdentityTarget},
+	})
+}
+
+func identityPath(spiffeID, trustDomain string) (string, error) {
+	parsed, err := url.Parse(spiffeID)
+	if err != nil || parsed.Scheme != "spiffe" || parsed.Host != trustDomain || !strings.HasPrefix(parsed.Path, "/ci/") {
+		return "", fmt.Errorf("invalid CI SPIFFE ID %q", spiffeID)
+	}
+	return strings.TrimPrefix(parsed.Path, "/ci/"), nil
+}
+
+func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error {
+	if err := b.validate(); err != nil {
+		return err
+	}
+	if executor == nil || executor.Name == "" {
+		return nil
+	}
+	if err := b.API.DeleteIdentityEntry(ctx, executor.Name); err != nil {
+		return fmt.Errorf("delete Pod identity entry: %w", err)
+	}
+	if err := b.API.DeletePod(ctx, b.Config.Namespace, executor.Name); err != nil {
+		return fmt.Errorf("delete assignment Pod: %w", err)
+	}
+	return nil
+}
+
+func (b Backend) validate() error {
+	if b.API == nil || b.Config.Namespace == "" || b.Config.Image == "" || b.Config.ServiceAccount == "" || b.Config.TrustDomain == "" || b.Config.SPIRECluster == "" || b.Config.SPIREClass == "" || b.Config.SPIREAgentID == "" || b.Config.ExecutorUID < 1 {
+		return errors.New("Pod API and complete executor/SPIRE configuration are required")
+	}
+	return nil
+}
+
+func executor(pod Pod) *taskworker.Executor {
+	return &taskworker.Executor{
+		Name:           pod.Name,
+		IdentityTarget: pod.UID,
+		Phase:          phase(pod.Phase),
+	}
+}
+
+func phase(value string) taskworker.Phase {
+	switch value {
+	case "Succeeded":
+		return taskworker.PhaseSucceeded
+	case "Failed":
+		return taskworker.PhaseFailed
+	case "Running":
+		return taskworker.PhaseRunning
+	default:
+		return taskworker.PhasePending
+	}
+}
+
+func clone(source map[string]string) map[string]string {
+	target := make(map[string]string, len(source))
+	for key, value := range source {
+		target[key] = value
+	}
+	return target
+}

internal/podbackend/backend_test.go

@@ -0,0 +1,153 @@
+package podbackend
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
+)
+
+type fakeAPI struct {
+	pods      []Pod
+	selector  string
+	created   PodManifest
+	entry     IdentityEntry
+	entryGone string
+	deleted   string
+	marked    map[string]string
+}
+
+func (a *fakeAPI) ListPods(_ context.Context, _ string, selector string) ([]Pod, error) {
+	a.selector = selector
+	return a.pods, nil
+}
+func (a *fakeAPI) CreatePod(_ context.Context, manifest PodManifest) (Pod, error) {
+	a.created = manifest
+	return Pod{Name: manifest.Name, Namespace: manifest.Namespace, UID: "pod-uid", Phase: "Pending"}, nil
+}
+func (a *fakeAPI) DeletePod(_ context.Context, _, name string) error {
+	a.deleted = name
+	return nil
+}
+func (a *fakeAPI) LabelPod(_ context.Context, _, _ string, labels map[string]string) error {
+	a.marked = labels
+	return nil
+}
+func (a *fakeAPI) EnsureIdentityEntry(_ context.Context, entry IdentityEntry) error {
+	a.entry = entry
+	return nil
+}
+func (a *fakeAPI) DeleteIdentityEntry(_ context.Context, name string) error {
+	a.entryGone = name
+	return nil
+}
+
+func backend(api API) Backend {
+	return Backend{API: api, Config: Config{
+		Namespace: "gitea-actions", Image: "zot/ci-executor:main",
+		ServiceAccount: "gitea-task-executor", ExecutorArgs: []string{"executor"},
+		TrustDomain: "ddupan.top", SPIRECluster: "homelab",
+		SPIREClass:   "spire-mgmt-spire",
+		SPIREAgentID: "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid",
+		ExecutorUID:  2000,
+	}}
+}
+
+func assignment() taskassignment.Assignment {
+	return taskassignment.Assignment{
+		ID: "gitea-task-42", Backend: taskassignment.BackendPod,
+		Task: &runnerv1.Task{Id: 42},
+		Identity: taskidentity.Identity{
+			Repository: "owner/repo", Task: "publish",
+			SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
+		},
+	}
+}
+
+func TestFindRecoversPodByAssignmentLabel(t *testing.T) {
+	api := &fakeAPI{pods: []Pod{{Name: "gitea-task-42", UID: "uid", Phase: "Running"}}}
+	executor, err := backend(api).Find(context.Background(), "gitea-task-42")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if api.selector != "ci.ddupan.top/assignment-id=gitea-task-42" || executor.Name != "gitea-task-42" || executor.IdentityTarget != "uid" || executor.Phase != taskworker.PhaseRunning {
+		t.Fatalf("selector=%q executor=%#v", api.selector, executor)
+	}
+}
+
+func TestCreateUsesDeterministicNameAndRecoveryMetadata(t *testing.T) {
+	api := &fakeAPI{}
+	metadata := taskworker.BackendMetadata(assignment())
+	executor, err := backend(api).Create(context.Background(), assignment(), taskworker.LaunchSpec{
+		Metadata: metadata, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if api.created.Name != "gitea-task-42" || api.created.Labels[assignmentLabel] != "gitea-task-42" {
+		t.Fatalf("manifest = %#v", api.created)
+	}
+	if api.created.Annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
+		t.Fatalf("annotations = %#v", api.created.Annotations)
+	}
+	if api.created.Environment["CI_RUNNER_CAPABILITY"] != "capability" {
+		t.Fatalf("environment = %#v", api.created.Environment)
+	}
+	if len(api.created.Args) != 1 || api.created.Args[0] != "executor" || executor.IdentityTarget != "pod-uid" {
+		t.Fatalf("args=%v executor=%#v", api.created.Args, executor)
+	}
+}
+
+func TestBindIdentityCreatesEntryPinnedToPodUID(t *testing.T) {
+	api := &fakeAPI{}
+	executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
+	if err := backend(api).BindIdentity(context.Background(), executor, assignment().Identity); err != nil {
+		t.Fatal(err)
+	}
+	if api.entry.Name != executor.Name || api.entry.SPIFFEID != assignment().Identity.SPIFFEID || api.entry.ParentID != "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid" || len(api.entry.Selectors) != 1 || api.entry.Selectors[0] != "k8s:pod-uid:pod-uid" {
+		t.Fatalf("entry=%#v", api.entry)
+	}
+}
+
+func TestDeleteRemovesIdentityBeforePod(t *testing.T) {
+	api := &fakeAPI{}
+	executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
+	if err := backend(api).Delete(context.Background(), executor); err != nil {
+		t.Fatal(err)
+	}
+	if api.entryGone != executor.Name || api.deleted != executor.Name {
+		t.Fatalf("entry=%q pod=%q", api.entryGone, api.deleted)
+	}
+}
+
+func TestTerminalMarkerAndCleanupUseBackendState(t *testing.T) {
+	api := &fakeAPI{pods: []Pod{
+		{Name: "running", UID: "running-uid", Phase: "Running"},
+		{Name: "finished", UID: "finished-uid", Phase: "Succeeded"},
+	}}
+	backend := backend(api)
+	if err := backend.MarkTerminal(context.Background(), "gitea-task-42"); err != nil {
+		t.Fatal(err)
+	}
+	if api.marked[terminalLabel] != "true" {
+		t.Fatalf("labels = %#v", api.marked)
+	}
+	cleaned, err := backend.CleanupTerminated(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if api.selector != terminalLabel+"=true" || cleaned != 1 || api.deleted != "finished" || api.entryGone != "finished" {
+		t.Fatalf("selector=%q cleaned=%d deleted=%q entry=%q", api.selector, cleaned, api.deleted, api.entryGone)
+	}
+}
+
+func TestFindRejectsDuplicatePods(t *testing.T) {
+	api := &fakeAPI{pods: []Pod{{Name: "one"}, {Name: "two"}}}
+	if _, err := backend(api).Find(context.Background(), "gitea-task-42"); err == nil {
+		t.Fatal("expected duplicate executor error")
+	}
+}

internal/podbackend/client.go

@@ -0,0 +1,178 @@
+package podbackend
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"reflect"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+var identityEntryResource = schema.GroupVersionResource{
+	Group: "spire.spiffe.io", Version: "v1alpha1", Resource: "clusterstaticentries",
+}
+
+// Client uses client-go's typed client for Pods and its dynamic client for the
+// SPIRE Operator CRD.
+type Client struct {
+	Kubernetes kubernetes.Interface
+	Dynamic    dynamic.Interface
+}
+
+func NewClient(config *rest.Config) (*Client, error) {
+	kubernetesClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("create Kubernetes client: %w", err)
+	}
+	dynamicClient, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("create Kubernetes dynamic client: %w", err)
+	}
+	return &Client{Kubernetes: kubernetesClient, Dynamic: dynamicClient}, nil
+}
+
+func NewInClusterClient() (*Client, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("load in-cluster Kubernetes config: %w", err)
+	}
+	return NewClient(config)
+}
+
+func (c *Client) ListPods(ctx context.Context, namespace, selector string) ([]Pod, error) {
+	list, err := c.Kubernetes.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector})
+	if err != nil {
+		return nil, err
+	}
+	pods := make([]Pod, 0, len(list.Items))
+	for _, item := range list.Items {
+		pods = append(pods, podFromKubernetes(item))
+	}
+	return pods, nil
+}
+
+func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, error) {
+	environment := make([]corev1.EnvVar, 0, len(manifest.Environment))
+	for name, value := range manifest.Environment {
+		environment = append(environment, corev1.EnvVar{Name: name, Value: value})
+	}
+	document := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: manifest.Name, Namespace: manifest.Namespace,
+			Labels: manifest.Labels, Annotations: manifest.Annotations,
+		},
+		Spec: corev1.PodSpec{
+			ServiceAccountName: manifest.ServiceAccount,
+			RestartPolicy:      corev1.RestartPolicyNever,
+			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,
+				}},
+			}},
+			Volumes: []corev1.Volume{{
+				Name: "spire-agent-socket",
+				VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
+					Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
+				}},
+			}},
+		},
+	}
+	created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
+	if err != nil {
+		return Pod{}, err
+	}
+	return podFromKubernetes(*created), nil
+}
+
+func (c *Client) DeletePod(ctx context.Context, namespace, name string) error {
+	policy := metav1.DeletePropagationBackground
+	err := c.Kubernetes.CoreV1().Pods(namespace).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	return err
+}
+
+func (c *Client) LabelPod(ctx context.Context, namespace, name string, labels map[string]string) error {
+	patch, err := json.Marshal(map[string]any{"metadata": map[string]any{"labels": labels}})
+	if err != nil {
+		return err
+	}
+	_, err = c.Kubernetes.CoreV1().Pods(namespace).Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{})
+	return err
+}
+
+func (c *Client) EnsureIdentityEntry(ctx context.Context, entry IdentityEntry) error {
+	resource := c.Dynamic.Resource(identityEntryResource)
+	existing, err := resource.Get(ctx, entry.Name, metav1.GetOptions{})
+	if err == nil {
+		existingSpec, _, nestedErr := unstructured.NestedMap(existing.Object, "spec")
+		if nestedErr != nil {
+			return nestedErr
+		}
+		if !reflect.DeepEqual(existingSpec, identityEntryObject(entry).Object["spec"]) {
+			return fmt.Errorf("identity entry %s exists with different selectors or SPIFFE ID", entry.Name)
+		}
+		return nil
+	}
+	if !apierrors.IsNotFound(err) {
+		return err
+	}
+	_, err = resource.Create(ctx, identityEntryObject(entry), metav1.CreateOptions{})
+	return err
+}
+
+func (c *Client) DeleteIdentityEntry(ctx context.Context, name string) error {
+	policy := metav1.DeletePropagationBackground
+	err := c.Dynamic.Resource(identityEntryResource).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	return err
+}
+
+func identityEntryObject(entry IdentityEntry) *unstructured.Unstructured {
+	selectors := make([]any, len(entry.Selectors))
+	for index, selector := range entry.Selectors {
+		selectors[index] = selector
+	}
+	return &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "spire.spiffe.io/v1alpha1",
+		"kind":       "ClusterStaticEntry",
+		"metadata": map[string]any{
+			"name": entry.Name, "labels": stringMap(entry.Labels),
+		},
+		"spec": map[string]any{
+			"className": entry.ClassName, "parentID": entry.ParentID,
+			"spiffeID": entry.SPIFFEID, "selectors": selectors,
+		},
+	}}
+}
+
+func podFromKubernetes(pod corev1.Pod) Pod {
+	return Pod{
+		Name: pod.Name, Namespace: pod.Namespace, UID: string(pod.UID),
+		Phase: string(pod.Status.Phase), Labels: pod.Labels, Annotations: pod.Annotations,
+	}
+}
+
+func boolPointer(value bool) *bool { return &value }
+
+func stringMap(values map[string]string) map[string]any {
+	result := make(map[string]any, len(values))
+	for key, value := range values {
+		result[key] = value
+	}
+	return result
+}

internal/podbackend/client_test.go

@@ -0,0 +1,82 @@
+package podbackend
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func testClient(objects ...runtime.Object) *Client {
+	return &Client{
+		Kubernetes: fake.NewSimpleClientset(objects...),
+		Dynamic:    dynamicfake.NewSimpleDynamicClient(runtime.NewScheme()),
+	}
+}
+
+func TestClientPodLifecycleUsesTypedClient(t *testing.T) {
+	client := testClient()
+	created, err := client.CreatePod(context.Background(), PodManifest{
+		Name: "gitea-task-42", Namespace: "gitea-actions",
+		Labels: map[string]string{assignmentLabel: "gitea-task-42"},
+		Image:  "zot/ci-executor:main", ServiceAccount: "gitea-task-executor",
+		Args: []string{"execute", "gitea-task-42"}, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pod, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	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 err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil {
+		t.Fatal(err)
+	}
+	pod, err = client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{})
+	if err != nil || pod.Labels[terminalLabel] != "true" {
+		t.Fatalf("terminal label pod=%#v err=%v", pod, err)
+	}
+	pod.UID = types.UID("pod-uid")
+	pod.Status.Phase = corev1.PodRunning
+	if _, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	pods, err := client.ListPods(context.Background(), "gitea-actions", assignmentLabel+"=gitea-task-42")
+	if err != nil || len(pods) != 1 || pods[0].UID != "pod-uid" || pods[0].Phase != "Running" {
+		t.Fatalf("pods=%#v err=%v", pods, err)
+	}
+	if err := client.DeletePod(context.Background(), "gitea-actions", created.Name); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestClientEnsuresIdempotentClusterStaticEntry(t *testing.T) {
+	client := testClient()
+	entry := IdentityEntry{
+		Name: "gitea-task-42", Labels: map[string]string{assignmentLabel: "gitea-task-42"},
+		ClassName: "spire-mgmt-spire",
+		ParentID:  "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/pod/pod-uid",
+		SPIFFEID:  "spiffe://ddupan.top/ci/owner/repo/publish",
+		Selectors: []string{"unix:uid:2000"},
+	}
+	if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
+		t.Fatal(err)
+	}
+	if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
+		t.Fatal(err)
+	}
+	if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
+		t.Fatal(err)
+	}
+	if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
+		t.Fatal(err)
+	}
+}

internal/podbackend/lifecycle.go

@@ -0,0 +1,32 @@
+package podbackend
+
+import (
+	"context"
+	"time"
+)
+
+// Lifecycle reconciles durable terminal markers into backend cleanup.
+type Lifecycle struct {
+	Backend  Backend
+	Interval time.Duration
+	OnError  func(error)
+}
+
+func (l Lifecycle) Run(ctx context.Context) error {
+	interval := l.Interval
+	if interval <= 0 {
+		interval = 2 * time.Second
+	}
+	for {
+		if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil {
+			l.OnError(err)
+		}
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil
+		case <-timer.C:
+		}
+	}
+}

internal/runnerbootstrap/bootstrap.go

@@ -0,0 +1,91 @@
+// Package runnerbootstrap configures an unmodified, one-shot Gitea Runner to
+// consume exactly the task assigned by the controller facade.
+package runnerbootstrap
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/url"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+const (
+	EnvAssignmentID = "CI_ASSIGNMENT_ID"
+	EnvCapability   = "CI_RUNNER_CAPABILITY"
+	EnvFacadeURL    = "CI_RUNNER_FACADE_URL"
+	EnvFacadeID     = "CI_RUNNER_FACADE_SPIFFE_ID"
+	EnvSPIFFEID     = "CI_SPIFFE_ID"
+	EnvBackend      = "CI_RUNNER_BACKEND"
+)
+
+// Bootstrap emits assignment-scoped launch configuration. FacadeURL is the
+// controller endpoint reached by the local SPIFFE proxy, not by Runner itself.
+type Bootstrap struct {
+	Capabilities    runnerfacade.Capabilities
+	FacadeURL       string
+	FacadeSPIFFEID  string
+	WorkloadAPIAddr string
+}
+
+func (b Bootstrap) Environment(assignment taskassignment.Assignment) (map[string]string, error) {
+	if assignment.ID == "" || assignment.Identity.SPIFFEID == "" || b.FacadeSPIFFEID == "" || b.WorkloadAPIAddr == "" {
+		return nil, errors.New("assignment ID and SPIFFE ID are required")
+	}
+	parsed, err := url.Parse(b.FacadeURL)
+	if err != nil || parsed.Scheme != "https" || parsed.Host == "" {
+		return nil, fmt.Errorf("runner facade URL must be an absolute https URL")
+	}
+	capability := b.Capabilities.Issue(assignment.ID)
+	if capability == "" {
+		return nil, errors.New("runner capability issuer is not configured")
+	}
+	return map[string]string{
+		EnvAssignmentID:          assignment.ID,
+		EnvCapability:            capability,
+		EnvFacadeURL:             b.FacadeURL,
+		EnvFacadeID:              b.FacadeSPIFFEID,
+		EnvSPIFFEID:              assignment.Identity.SPIFFEID,
+		EnvBackend:               string(assignment.Backend),
+		"SPIFFE_ENDPOINT_SOCKET": b.WorkloadAPIAddr,
+	}, nil
+}
+
+// Registration mirrors Gitea Runner v3.5.0's registration file schema. ID is
+// intentionally zero: the facade authenticates UUID and token and never uses
+// the server-issued runner database ID.
+type Registration struct {
+	Warning   string   `json:"WARNING"`
+	ID        int64    `json:"id"`
+	UUID      string   `json:"uuid"`
+	Name      string   `json:"name"`
+	Token     string   `json:"token"`
+	Address   string   `json:"address"`
+	Labels    []string `json:"labels"`
+	Ephemeral bool     `json:"ephemeral"`
+}
+
+func RegistrationJSON(assignmentID, capability, localProxyURL string, backend taskassignment.Backend) ([]byte, error) {
+	if assignmentID == "" || capability == "" {
+		return nil, errors.New("assignment ID and runner capability are required")
+	}
+	parsed, err := url.Parse(localProxyURL)
+	if err != nil || parsed.Scheme != "http" || parsed.Host == "" {
+		return nil, errors.New("local runner proxy URL must be an absolute http URL")
+	}
+	if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
+		return nil, fmt.Errorf("unsupported runner backend %q", backend)
+	}
+	registration := Registration{
+		Warning: "Generated for one preassigned task by gitea-dynamic-runner.",
+		UUID:    assignmentID, Name: assignmentID, Token: capability,
+		Address: localProxyURL, Labels: []string{"self-hosted", string(backend)}, Ephemeral: true,
+	}
+	data, err := json.MarshalIndent(registration, "", "  ")
+	if err != nil {
+		return nil, err
+	}
+	return append(data, '\n'), nil
+}

internal/runnerbootstrap/bootstrap_test.go

@@ -0,0 +1,78 @@
+package runnerbootstrap
+
+import (
+	"encoding/json"
+	"testing"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+)
+
+func testBootstrap(t *testing.T) Bootstrap {
+	t.Helper()
+	capabilities, err := runnerfacade.NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return Bootstrap{
+		Capabilities:    capabilities,
+		FacadeURL:       "https://runner-facade.gitea-actions.svc:8443",
+		FacadeSPIFFEID:  "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner",
+		WorkloadAPIAddr: "unix:///run/spire/agent-sockets/spire-agent.sock",
+	}
+}
+
+func testAssignment() taskassignment.Assignment {
+	return taskassignment.Assignment{
+		ID: "gitea-task-42", Backend: taskassignment.BackendPod,
+		Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
+	}
+}
+
+func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) {
+	bootstrap := testBootstrap(t)
+	first, err := bootstrap.Environment(testAssignment())
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := bootstrap.Environment(testAssignment())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first[EnvCapability] == "" || first[EnvCapability] != second[EnvCapability] {
+		t.Fatalf("capabilities = %q, %q", first[EnvCapability], second[EnvCapability])
+	}
+	if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID {
+		t.Fatalf("environment = %#v", first)
+	}
+	if first[EnvBackend] != "pod" || first[EnvFacadeID] == "" {
+		t.Fatalf("environment = %#v", first)
+	}
+	if first["SPIFFE_ENDPOINT_SOCKET"] != "unix:///run/spire/agent-sockets/spire-agent.sock" {
+		t.Fatalf("environment = %#v", first)
+	}
+}
+
+func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) {
+	data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.BackendVM)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var registration Registration
+	if err := json.Unmarshal(data, &registration); err != nil {
+		t.Fatal(err)
+	}
+	if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral {
+		t.Fatalf("registration = %#v", registration)
+	}
+	if len(registration.Labels) != 2 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" {
+		t.Fatalf("labels = %#v", registration.Labels)
+	}
+}
+
+func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) {
+	if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.BackendPod); err == nil {
+		t.Fatal("expected local proxy URL validation error")
+	}
+}

internal/runnerbootstrap/executor.go

@@ -0,0 +1,153 @@
+package runnerbootstrap
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"time"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type ExecutorConfig struct {
+	AssignmentID    string
+	Capability      string
+	Backend         taskassignment.Backend
+	FacadeURL       string
+	FacadeSPIFFEID  string
+	WorkloadAPIAddr string
+	RunnerBinary    string
+	ListenAddress   string
+	WorkDir         string
+	Stdout          *os.File
+	Stderr          *os.File
+}
+
+// RunExecutor runs the SPIFFE proxy and one unmodified official Runner process.
+// The generated registration file exists only in the executor's temporary
+// work directory and the runner exits after its preassigned task.
+func RunExecutor(ctx context.Context, config ExecutorConfig) error {
+	if config.RunnerBinary == "" {
+		config.RunnerBinary = "gitea-runner"
+	}
+	if config.ListenAddress == "" {
+		config.ListenAddress = "127.0.0.1:0"
+	}
+	listener, err := net.Listen("tcp", config.ListenAddress)
+	if err != nil {
+		return fmt.Errorf("listen for local runner proxy: %w", err)
+	}
+	defer listener.Close()
+	address, ok := listener.Addr().(*net.TCPAddr)
+	if !ok || !address.IP.IsLoopback() {
+		return errors.New("runner proxy must listen on a loopback address")
+	}
+
+	proxy, err := NewProxy(ctx, config.FacadeURL, config.FacadeSPIFFEID, config.WorkloadAPIAddr)
+	if err != nil {
+		return err
+	}
+	defer proxy.Close()
+
+	workDir := config.WorkDir
+	removeWorkDir := false
+	if workDir == "" {
+		workDir, err = os.MkdirTemp("", "gitea-task-runner-")
+		if err != nil {
+			return fmt.Errorf("create runner work directory: %w", err)
+		}
+		removeWorkDir = true
+	}
+	if removeWorkDir {
+		defer os.RemoveAll(workDir)
+	}
+	registration, err := RegistrationJSON(
+		config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend,
+	)
+	if err != nil {
+		return err
+	}
+	if err := os.WriteFile(filepath.Join(workDir, ".runner"), registration, 0o600); err != nil {
+		return fmt.Errorf("write one-shot runner registration: %w", err)
+	}
+
+	server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second}
+	serverErrors := make(chan error, 1)
+	go func() { serverErrors <- server.Serve(listener) }()
+	readyContext, cancelReady := context.WithTimeout(ctx, 2*time.Minute)
+	readyErr := waitForFacade(readyContext, "http://"+listener.Addr().String())
+	cancelReady()
+	if readyErr != nil {
+		shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		shutdownErr := server.Shutdown(shutdownContext)
+		cancel()
+		serverErr := <-serverErrors
+		if errors.Is(serverErr, http.ErrServerClosed) {
+			serverErr = nil
+		}
+		return errors.Join(readyErr, shutdownErr, serverErr)
+	}
+
+	command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
+	command.Dir = workDir
+	command.Stdout = config.Stdout
+	command.Stderr = config.Stderr
+	runnerErr := command.Run()
+	shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	shutdownErr := server.Shutdown(shutdownContext)
+	cancel()
+	serverErr := <-serverErrors
+	if errors.Is(serverErr, http.ErrServerClosed) {
+		serverErr = nil
+	}
+	return errors.Join(runnerErr, shutdownErr, serverErr)
+}
+
+func waitForFacade(ctx context.Context, endpoint string) error {
+	client := &http.Client{Timeout: 2 * time.Second}
+	ticker := time.NewTicker(250 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+		if err != nil {
+			return fmt.Errorf("create facade readiness request: %w", err)
+		}
+		response, requestErr := client.Do(request)
+		if requestErr == nil {
+			_ = response.Body.Close()
+			if response.StatusCode < http.StatusInternalServerError {
+				return nil
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("wait for runner facade: %w", ctx.Err())
+		case <-ticker.C:
+		}
+	}
+}
+
+// ExecutorConfigFromEnvironment reads the non-secret image configuration and
+// the assignment-scoped values injected by the backend. The Workload API
+// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
+func ExecutorConfigFromEnvironment() (ExecutorConfig, error) {
+	backend := taskassignment.Backend(os.Getenv(EnvBackend))
+	if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
+		return ExecutorConfig{}, fmt.Errorf("invalid %s %q", EnvBackend, backend)
+	}
+	config := ExecutorConfig{
+		AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability),
+		Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID),
+		RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), ListenAddress: "127.0.0.1:0",
+		Stdout: os.Stdout, Stderr: os.Stderr,
+	}
+	if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
+		return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required")
+	}
+	return config, nil
+}

internal/runnerbootstrap/executor_test.go

@@ -0,0 +1,44 @@
+package runnerbootstrap
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestWaitForFacadeRetriesTransientGatewayFailure(t *testing.T) {
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		if requests.Add(1) < 3 {
+			writer.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		writer.WriteHeader(http.StatusNotFound)
+	}))
+	defer server.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	if err := waitForFacade(ctx, server.URL); err != nil {
+		t.Fatal(err)
+	}
+	if requests.Load() != 3 {
+		t.Fatalf("requests = %d", requests.Load())
+	}
+}
+
+func TestWaitForFacadeStopsWithContext(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		writer.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer server.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	if err := waitForFacade(ctx, server.URL); err == nil {
+		t.Fatal("expected readiness timeout")
+	}
+}

internal/runnerbootstrap/proxy.go

@@ -0,0 +1,56 @@
+package runnerbootstrap
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+
+	"github.com/spiffe/go-spiffe/v2/spiffeid"
+	"github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig"
+	"github.com/spiffe/go-spiffe/v2/workloadapi"
+)
+
+type Proxy struct {
+	Handler http.Handler
+	source  *workloadapi.X509Source
+}
+
+// NewProxy obtains rotating X509-SVIDs from the Workload API and authorizes
+// one exact controller identity. The official Runner talks plain HTTP only to
+// this executor-local handler.
+func NewProxy(ctx context.Context, facadeURL, facadeSPIFFEID, workloadAPIAddr string) (*Proxy, error) {
+	target, err := url.Parse(facadeURL)
+	if err != nil || target.Scheme != "https" || target.Host == "" {
+		return nil, fmt.Errorf("runner facade URL must be an absolute https URL")
+	}
+	serverID, err := spiffeid.FromString(facadeSPIFFEID)
+	if err != nil {
+		return nil, fmt.Errorf("parse runner facade SPIFFE ID: %w", err)
+	}
+	options := []workloadapi.X509SourceOption{}
+	if workloadAPIAddr != "" {
+		options = append(options, workloadapi.WithClientOptions(workloadapi.WithAddr(workloadAPIAddr)))
+	}
+	source, err := workloadapi.NewX509Source(ctx, options...)
+	if err != nil {
+		return nil, fmt.Errorf("open SPIFFE Workload API X509 source: %w", err)
+	}
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	transport.TLSClientConfig = tlsconfig.MTLSClientConfig(source, source, tlsconfig.AuthorizeID(serverID))
+	return &Proxy{Handler: NewProxyHandler(target, transport), source: source}, nil
+}
+
+func NewProxyHandler(target *url.URL, transport http.RoundTripper) http.Handler {
+	proxy := httputil.NewSingleHostReverseProxy(target)
+	proxy.Transport = transport
+	return proxy
+}
+
+func (p *Proxy) Close() error {
+	if p == nil || p.source == nil {
+		return nil
+	}
+	return p.source.Close()
+}

internal/runnerfacade/capability.go

@@ -0,0 +1,36 @@
+package runnerfacade
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"errors"
+)
+
+// Capabilities are deterministic per assignment so controller restarts do not
+// require a per-task token database.
+type Capabilities struct{ key []byte }
+
+func NewCapabilities(key []byte) (Capabilities, error) {
+	if len(key) < 32 {
+		return Capabilities{}, errors.New("runner facade capability key must be at least 32 bytes")
+	}
+	return Capabilities{key: append([]byte(nil), key...)}, nil
+}
+
+func (c Capabilities) Issue(assignmentID string) string {
+	if len(c.key) < 32 || assignmentID == "" {
+		return ""
+	}
+	mac := hmac.New(sha256.New, c.key)
+	_, _ = mac.Write([]byte("gitea-runner-assignment\x00" + assignmentID))
+	return base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
+}
+
+func (c Capabilities) Verify(assignmentID, token string) bool {
+	want := c.Issue(assignmentID)
+	if want == "" || token == "" {
+		return false
+	}
+	return hmac.Equal([]byte(want), []byte(token))
+}

internal/runnerfacade/facade.go

@@ -0,0 +1,145 @@
+// Package runnerfacade presents pre-assigned tasks to unmodified Gitea Runner binaries.
+package runnerfacade
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type identityKey struct{}
+
+func WithSPIFFEID(ctx context.Context, id string) context.Context {
+	return context.WithValue(ctx, identityKey{}, id)
+}
+
+// SPIFFEMiddleware extracts the authenticated workload identity from the mTLS
+// peer certificate. TLS verification itself is configured by the server with
+// go-spiffe; this layer only passes the verified ID into Connect handlers.
+func SPIFFEMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		if request.TLS == nil || len(request.TLS.PeerCertificates) == 0 {
+			http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
+			return
+		}
+		var spiffeID string
+		for _, uri := range request.TLS.PeerCertificates[0].URIs {
+			if uri.Scheme == "spiffe" {
+				if spiffeID != "" {
+					http.Error(response, "multiple client SPIFFE identities", http.StatusUnauthorized)
+					return
+				}
+				spiffeID = uri.String()
+			}
+		}
+		if spiffeID == "" {
+			http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
+			return
+		}
+		next.ServeHTTP(response, request.WithContext(WithSPIFFEID(request.Context(), spiffeID)))
+	})
+}
+
+type Upstream interface {
+	UpdateTask(context.Context, *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error)
+	UpdateLog(context.Context, *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error)
+}
+
+type Facade struct {
+	runnerv1connect.UnimplementedRunnerServiceHandler
+	Registry     *Registry
+	Capabilities Capabilities
+	Upstream     Upstream
+	OnTerminal   func(context.Context, taskassignment.Assignment) error
+}
+
+func (f *Facade) Handler() (string, http.Handler) {
+	return runnerv1connect.NewRunnerServiceHandler(f)
+}
+
+func (f *Facade) Register(context.Context, *connect.Request[runnerv1.RegisterRequest]) (*connect.Response[runnerv1.RegisterResponse], error) {
+	return nil, connect.NewError(connect.CodePermissionDenied, errors.New("executor registration is disabled"))
+}
+
+func (f *Facade) Declare(ctx context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
+	assignmentID, _, err := f.authenticate(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	return connect.NewResponse(&runnerv1.DeclareResponse{Runner: &runnerv1.Runner{
+		Uuid: assignmentID, Name: assignmentID, Status: runnerv1.RunnerStatus_RUNNER_STATUS_IDLE,
+		Version: request.Msg.GetVersion(), Labels: append([]string(nil), request.Msg.GetLabels()...), Ephemeral: true,
+	}}), nil
+}
+
+func (f *Facade) FetchTask(ctx context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
+	assignmentID, spiffeID, err := f.authenticate(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	assignment, err := f.Registry.Claim(assignmentID, spiffeID)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeFailedPrecondition, err)
+	}
+	return connect.NewResponse(&runnerv1.FetchTaskResponse{Task: assignment.Task}), nil
+}
+
+func (f *Facade) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
+	assignment, err := f.authorizeClaimed(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	if request.Msg.GetState().GetId() != assignment.Task.GetId() {
+		return nil, connect.NewError(connect.CodePermissionDenied, errors.New("task update does not match assignment"))
+	}
+	response, err := f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg))
+	if err == nil && request.Msg.GetState().GetResult() != runnerv1.Result_RESULT_UNSPECIFIED && f.OnTerminal != nil {
+		if terminalErr := f.OnTerminal(ctx, assignment); terminalErr != nil {
+			return nil, connect.NewError(connect.CodeUnavailable, terminalErr)
+		}
+	}
+	return response, err
+}
+
+func (f *Facade) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
+	assignment, err := f.authorizeClaimed(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	if request.Msg.GetTaskId() != assignment.Task.GetId() {
+		return nil, connect.NewError(connect.CodePermissionDenied, errors.New("log update does not match assignment"))
+	}
+	return f.Upstream.UpdateLog(ctx, connect.NewRequest(request.Msg))
+}
+
+func (f *Facade) authorizeClaimed(ctx context.Context, request connect.AnyRequest) (taskassignment.Assignment, error) {
+	assignmentID, spiffeID, err := f.authenticate(ctx, request)
+	if err != nil {
+		return taskassignment.Assignment{}, err
+	}
+	assignment, err := f.Registry.Resolve(assignmentID, spiffeID)
+	if err != nil {
+		return taskassignment.Assignment{}, connect.NewError(connect.CodePermissionDenied, err)
+	}
+	return assignment, nil
+}
+
+func (f *Facade) authenticate(ctx context.Context, request connect.AnyRequest) (string, string, error) {
+	if f.Registry == nil || f.Upstream == nil {
+		return "", "", connect.NewError(connect.CodeInternal, errors.New("runner facade is not configured"))
+	}
+	assignmentID := request.Header().Get(protocol.UUIDHeader)
+	token := request.Header().Get(protocol.TokenHeader)
+	spiffeID, _ := ctx.Value(identityKey{}).(string)
+	if assignmentID == "" || spiffeID == "" || !f.Capabilities.Verify(assignmentID, token) {
+		return "", "", connect.NewError(connect.CodeUnauthenticated, errors.New("invalid executor identity or capability"))
+	}
+	return assignmentID, spiffeID, nil
+}

internal/runnerfacade/facade_test.go

@@ -0,0 +1,239 @@
+package runnerfacade
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type fakeUpstream struct {
+	taskUpdates int
+	logUpdates  int
+}
+
+func (u *fakeUpstream) UpdateTask(_ context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
+	u.taskUpdates++
+	return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: request.Msg.State}), nil
+}
+func (u *fakeUpstream) UpdateLog(_ context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
+	u.logUpdates++
+	return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: request.Msg.Index + int64(len(request.Msg.Rows))}), nil
+}
+
+func facadeAssignment(t *testing.T) taskassignment.Assignment {
+	t.Helper()
+	fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assignment, err := taskassignment.New(&runnerv1.Task{
+		Id: 42, Context: fields,
+		WorkflowPayload: []byte("jobs:\n  publish:\n    runs-on: [self-hosted, pod]\n    steps: []\n"),
+	}, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return assignment
+}
+
+func testFacade(t *testing.T) (*Facade, taskassignment.Assignment, string) {
+	t.Helper()
+	capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assignment := facadeAssignment(t)
+	registry := NewRegistry()
+	if _, err := registry.Offer(assignment); err != nil {
+		t.Fatal(err)
+	}
+	return &Facade{Registry: registry, Capabilities: capabilities, Upstream: &fakeUpstream{}}, assignment, capabilities.Issue(assignment.ID)
+}
+
+func authenticatedRequest[T any](message *T, assignmentID, token string) *connect.Request[T] {
+	request := connect.NewRequest(message)
+	request.Header().Set(protocol.UUIDHeader, assignmentID)
+	request.Header().Set(protocol.TokenHeader, token)
+	return request
+}
+
+func TestFacadeReturnsOnlyPreassignedTaskAndSignalsClaim(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
+	response, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.Msg.GetTask().GetId() != assignment.Task.GetId() {
+		t.Fatalf("task = %#v", response.Msg.GetTask())
+	}
+	if err := facade.Registry.WaitClaimed(context.Background(), assignment.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition {
+		t.Fatalf("second FetchTask error = %v", err)
+	}
+}
+
+func TestFacadeDeclareReturnsOfficialRunnerMetadata(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
+	response, err := facade.Declare(ctx, authenticatedRequest(&runnerv1.DeclareRequest{
+		Version: "v3.5.0", Labels: []string{"self-hosted", "pod"},
+	}, assignment.ID, token))
+	if err != nil {
+		t.Fatal(err)
+	}
+	runner := response.Msg.GetRunner()
+	if runner.GetUuid() != assignment.ID || runner.GetName() != assignment.ID || runner.GetVersion() != "v3.5.0" || !runner.GetEphemeral() {
+		t.Fatalf("runner = %#v", runner)
+	}
+	if len(runner.GetLabels()) != 2 || runner.GetLabels()[0] != "self-hosted" || runner.GetLabels()[1] != "pod" {
+		t.Fatalf("labels = %#v", runner.GetLabels())
+	}
+}
+
+func TestAPIHandlerMatchesOfficialRunnerBasePath(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	identityURL, err := url.Parse(assignment.Identity.SPIFFEID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	handler := APIHandler(facade)
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		request.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{{URIs: []*url.URL{identityURL}}}}
+		handler.ServeHTTP(response, request)
+	}))
+	defer server.Close()
+	client := runnerv1connect.NewRunnerServiceClient(server.Client(), server.URL+APIBasePath)
+	request := authenticatedRequest(&runnerv1.DeclareRequest{
+		Version: "v3.5.0", Labels: []string{"self-hosted", "pod"},
+	}, assignment.ID, token)
+	response, err := client.Declare(context.Background(), request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.Msg.GetRunner().GetUuid() != assignment.ID {
+		t.Fatalf("runner = %#v", response.Msg.GetRunner())
+	}
+}
+
+func TestHandlerProxiesRepositoryTrafficToGitea(t *testing.T) {
+	var upstream *httptest.Server
+	upstream = httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		upstreamHost := strings.TrimPrefix(upstream.URL, "http://")
+		if request.URL.Path != "/owner/repo/info/refs" || request.Header.Get("Authorization") != "Basic checkout-token" || request.Host != upstreamHost {
+			t.Errorf("request path=%q authorization=%q host=%q", request.URL.Path, request.Header.Get("Authorization"), request.Host)
+			response.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		response.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	facade, _, _ := testFacade(t)
+	handler, err := Handler(facade, upstream.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := httptest.NewRequest(http.MethodGet, "http://facade/owner/repo/info/refs", nil)
+	request.Header.Set("Authorization", "Basic checkout-token")
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, request)
+	if response.Code != http.StatusOK {
+		t.Fatalf("status = %d", response.Code)
+	}
+}
+
+func TestFacadeRejectsWrongIdentityOrCapability(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	wrongIdentity := WithSPIFFEID(context.Background(), "spiffe://ddupan.top/ci/owner/repo/other")
+	if _, err := facade.FetchTask(wrongIdentity, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition {
+		t.Fatalf("wrong identity error = %v", err)
+	}
+	ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
+	if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, "wrong")); connect.CodeOf(err) != connect.CodeUnauthenticated {
+		t.Fatalf("wrong capability error = %v", err)
+	}
+}
+
+func TestFacadeForwardsOnlyMatchingTaskAndLogUpdates(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
+	if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{State: &runnerv1.TaskState{Id: 42}}, assignment.ID, token)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 42}, assignment.ID, token)); err != nil {
+		t.Fatal(err)
+	}
+	upstream := facade.Upstream.(*fakeUpstream)
+	if upstream.taskUpdates != 1 || upstream.logUpdates != 1 {
+		t.Fatalf("task updates=%d log updates=%d", upstream.taskUpdates, upstream.logUpdates)
+	}
+	if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 99}, assignment.ID, token)); connect.CodeOf(err) != connect.CodePermissionDenied {
+		t.Fatalf("mismatched log error = %v", err)
+	}
+}
+
+func TestFacadeSignalsTerminalTaskAfterUpstreamAcceptsIt(t *testing.T) {
+	facade, assignment, token := testFacade(t)
+	ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
+	if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil {
+		t.Fatal(err)
+	}
+	completed := 0
+	facade.OnTerminal = func(_ context.Context, got taskassignment.Assignment) error {
+		if got.ID != assignment.ID {
+			t.Fatalf("terminal assignment = %s", got.ID)
+		}
+		completed++
+		return nil
+	}
+	if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{
+		State: &runnerv1.TaskState{Id: 42, Result: runnerv1.Result_RESULT_SUCCESS},
+	}, assignment.ID, token)); err != nil {
+		t.Fatal(err)
+	}
+	if completed != 1 {
+		t.Fatalf("terminal notifications = %d", completed)
+	}
+}
+
+func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) {
+	capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	token := capabilities.Issue("gitea-task-42")
+	if !capabilities.Verify("gitea-task-42", token) || capabilities.Verify("gitea-task-43", token) {
+		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

@@ -0,0 +1,110 @@
+package runnerfacade
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type claim struct {
+	assignment taskassignment.Assignment
+	claimed    bool
+	ready      chan struct{}
+}
+
+// Registry holds pending task payloads and an active authorization cache.
+// Pending entries are rebuilt by JetStream redelivery; active authorization is
+// reconstructable from backend metadata and is not an independent state store.
+type Registry struct {
+	mu     sync.Mutex
+	claims map[string]*claim
+}
+
+func NewRegistry() *Registry { return &Registry{claims: make(map[string]*claim)} }
+
+func (r *Registry) Offer(assignment taskassignment.Assignment) (<-chan struct{}, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if existing := r.claims[assignment.ID]; existing != nil {
+		if existing.assignment.Identity != assignment.Identity || existing.assignment.Task.GetId() != assignment.Task.GetId() {
+			return nil, fmt.Errorf("assignment %s was offered with different task data", assignment.ID)
+		}
+		return existing.ready, nil
+	}
+	entry := &claim{assignment: assignment, ready: make(chan struct{})}
+	r.claims[assignment.ID] = entry
+	return entry.ready, nil
+}
+
+func (r *Registry) Claim(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	entry := r.claims[assignmentID]
+	if entry == nil {
+		return taskassignment.Assignment{}, errors.New("assignment is not pending")
+	}
+	if entry.assignment.Identity.SPIFFEID != spiffeID {
+		return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
+	}
+	if entry.claimed {
+		return taskassignment.Assignment{}, errors.New("assignment was already claimed")
+	}
+	entry.claimed = true
+	close(entry.ready)
+	return entry.assignment, nil
+}
+
+func (r *Registry) Resolve(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	entry := r.claims[assignmentID]
+	if entry == nil || !entry.claimed {
+		return taskassignment.Assignment{}, errors.New("assignment is not claimed")
+	}
+	if entry.assignment.Identity.SPIFFEID != spiffeID {
+		return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
+	}
+	return entry.assignment, nil
+}
+
+func (r *Registry) WaitClaimed(ctx context.Context, assignmentID string) error {
+	r.mu.Lock()
+	entry := r.claims[assignmentID]
+	r.mu.Unlock()
+	if entry == nil {
+		return errors.New("assignment is not pending")
+	}
+	select {
+	case <-entry.ready:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (r *Registry) Remove(assignmentID string) {
+	r.mu.Lock()
+	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/runnerfacade/server.go

@@ -0,0 +1,113 @@
+package runnerfacade
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"time"
+
+	"github.com/spiffe/go-spiffe/v2/spiffeid"
+	"github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig"
+	"github.com/spiffe/go-spiffe/v2/workloadapi"
+)
+
+const APIBasePath = "/api/actions"
+
+// APIHandler exposes the facade at the base path used by the official Runner.
+// SPIFFE middleware runs after the TLS listener has authenticated the peer.
+func APIHandler(facade *Facade) http.Handler {
+	path, handler := facade.Handler()
+	mux := http.NewServeMux()
+	mux.Handle(APIBasePath+path, http.StripPrefix(APIBasePath, SPIFFEMiddleware(handler)))
+	return mux
+}
+
+// Handler keeps RunnerService calls inside the authenticated facade while
+// forwarding repository and artifact HTTP traffic to the real Gitea server.
+// Official Runner derives checkout URLs from its registration instance URL,
+// which intentionally points at the executor-local SPIFFE proxy.
+func Handler(facade *Facade, upstreamURL string) (http.Handler, error) {
+	target, err := url.Parse(upstreamURL)
+	if err != nil || (target.Scheme != "http" && target.Scheme != "https") || target.Host == "" {
+		return nil, errors.New("Gitea upstream must be an absolute HTTP URL")
+	}
+	path, service := facade.Handler()
+	mux := http.NewServeMux()
+	mux.Handle(APIBasePath+path, http.StripPrefix(APIBasePath, SPIFFEMiddleware(service)))
+	proxy := httputil.NewSingleHostReverseProxy(target)
+	director := proxy.Director
+	proxy.Director = func(request *http.Request) {
+		director(request)
+		request.Host = target.Host
+	}
+	mux.Handle("/", proxy)
+	return mux, nil
+}
+
+type Server struct {
+	Facade          *Facade
+	ListenAddress   string
+	TrustDomain     string
+	WorkloadAPIAddr string
+	UpstreamURL     string
+}
+
+// Run serves the RunnerService facade with workload-to-workload mTLS. Any
+// identity in the local trust domain may complete TLS; the facade then requires
+// the exact logical task identity stored in its assignment registry.
+func (s Server) Run(ctx context.Context) error {
+	if s.Facade == nil || s.ListenAddress == "" || s.TrustDomain == "" || s.UpstreamURL == "" {
+		return errors.New("runner facade, listen address, trust domain, and Gitea upstream are required")
+	}
+	handler, err := Handler(s.Facade, s.UpstreamURL)
+	if err != nil {
+		return err
+	}
+	trustDomain, err := spiffeid.TrustDomainFromString(s.TrustDomain)
+	if err != nil {
+		return fmt.Errorf("parse facade trust domain: %w", err)
+	}
+	options := []workloadapi.X509SourceOption{}
+	if s.WorkloadAPIAddr != "" {
+		options = append(options, workloadapi.WithClientOptions(workloadapi.WithAddr(s.WorkloadAPIAddr)))
+	}
+	source, err := workloadapi.NewX509Source(ctx, options...)
+	if err != nil {
+		return fmt.Errorf("open facade SPIFFE Workload API X509 source: %w", err)
+	}
+	defer source.Close()
+
+	listener, err := net.Listen("tcp", s.ListenAddress)
+	if err != nil {
+		return fmt.Errorf("listen for runner facade: %w", err)
+	}
+	defer listener.Close()
+	tlsListener := tls.NewListener(listener, tlsconfig.MTLSServerConfig(
+		source, source, tlsconfig.AuthorizeMemberOf(trustDomain),
+	))
+	httpServer := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second}
+	serverErrors := make(chan error, 1)
+	go func() { serverErrors <- httpServer.Serve(tlsListener) }()
+
+	select {
+	case <-ctx.Done():
+		shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		shutdownErr := httpServer.Shutdown(shutdownContext)
+		cancel()
+		serverErr := <-serverErrors
+		if errors.Is(serverErr, http.ErrServerClosed) {
+			serverErr = nil
+		}
+		return errors.Join(shutdownErr, serverErr)
+	case err := <-serverErrors:
+		if errors.Is(err, http.ErrServerClosed) {
+			return nil
+		}
+		return fmt.Errorf("serve runner facade: %w", err)
+	}
+}

internal/taskassignment/assignment.go

@@ -0,0 +1,152 @@
+// Package taskassignment defines the durable handoff between the scheduler and workers.
+package taskassignment
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"slices"
+	"strconv"
+
+	"gitea.dev/actionslib/pkg/model"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/proto"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+)
+
+const wireVersion = 1
+
+type Backend string
+
+const (
+	BackendPod Backend = "pod"
+	BackendVM  Backend = "vm"
+)
+
+// Assignment is the only document persisted in the handoff queue.
+type Assignment struct {
+	ID       string
+	Backend  Backend
+	Task     *runnerv1.Task
+	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"`
+	Backend  Backend               `json:"backend"`
+	Task     []byte                `json:"task"`
+	Identity taskidentity.Identity `json:"identity"`
+}
+
+// New derives all trusted assignment fields from the task fetched from Gitea.
+func New(task *runnerv1.Task, trustDomain string) (Assignment, error) {
+	if task == nil || task.GetId() <= 0 {
+		return Assignment{}, errors.New("positive Gitea task ID is required")
+	}
+	identity, err := taskidentity.FromTask(task, trustDomain)
+	if err != nil {
+		return Assignment{}, err
+	}
+	backend, err := backendFromTask(task)
+	if err != nil {
+		return Assignment{}, err
+	}
+	return Assignment{
+		ID:       fmt.Sprintf("gitea-task-%d", task.GetId()),
+		Backend:  backend,
+		Task:     task,
+		Identity: identity,
+	}, nil
+}
+
+func backendFromTask(task *runnerv1.Task) (Backend, error) {
+	workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload()))
+	if err != nil {
+		return "", fmt.Errorf("parse task workflow for backend: %w", err)
+	}
+	jobIDs := workflow.GetJobIDs()
+	if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil {
+		return "", fmt.Errorf("task workflow must contain exactly one non-empty job")
+	}
+	labels := workflow.GetJob(jobIDs[0]).RunsOnLabels()
+	if !slices.Contains(labels, "self-hosted") {
+		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)) || 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)
+	}
+	if hasPod {
+		return BackendPod, nil
+	}
+	return BackendVM, nil
+}
+
+// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task.
+func Marshal(assignment Assignment) ([]byte, error) {
+	if assignment.Task == nil {
+		return nil, errors.New("assignment task is required")
+	}
+	task, err := proto.Marshal(assignment.Task)
+	if err != nil {
+		return nil, fmt.Errorf("marshal Gitea task: %w", err)
+	}
+	return json.Marshal(envelope{
+		Version: wireVersion,
+		ID:      assignment.ID, Backend: assignment.Backend,
+		Task: task, Identity: assignment.Identity,
+	})
+}
+
+// Unmarshal re-derives trusted fields instead of trusting duplicated queue metadata.
+func Unmarshal(data []byte, trustDomain string) (Assignment, error) {
+	var wire envelope
+	if err := json.Unmarshal(data, &wire); err != nil {
+		return Assignment{}, fmt.Errorf("decode assignment: %w", err)
+	}
+	if wire.Version != wireVersion {
+		return Assignment{}, fmt.Errorf("unsupported assignment version %d", wire.Version)
+	}
+	task := new(runnerv1.Task)
+	if err := proto.Unmarshal(wire.Task, task); err != nil {
+		return Assignment{}, fmt.Errorf("unmarshal Gitea task: %w", err)
+	}
+	canonical, err := New(task, trustDomain)
+	if err != nil {
+		return Assignment{}, err
+	}
+	if wire.ID != canonical.ID || wire.Backend != canonical.Backend || wire.Identity != canonical.Identity {
+		return Assignment{}, errors.New("assignment metadata does not match its Gitea task")
+	}
+	return canonical, nil
+}

internal/taskassignment/assignment_test.go

@@ -0,0 +1,91 @@
+package taskassignment
+
+import (
+	"bytes"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func task(t *testing.T, labels string) *runnerv1.Task {
+	t.Helper()
+	context, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &runnerv1.Task{
+		Id:              42,
+		Context:         context,
+		WorkflowPayload: []byte("jobs:\n  publish:\n    runs-on: " + labels + "\n    steps: []\n"),
+	}
+}
+
+func TestNewSelectsBackendFromRunsOn(t *testing.T) {
+	for _, test := range []struct {
+		labels  string
+		backend Backend
+	}{
+		{"[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 {
+			t.Fatal(err)
+		}
+		if assignment.Backend != test.backend || assignment.ID != "gitea-task-42" {
+			t.Fatalf("assignment = %#v", assignment)
+		}
+	}
+}
+
+func TestNewRejectsAmbiguousBackend(t *testing.T) {
+	for _, labels := range []string{
+		"[self-hosted]",
+		"[self-hosted, pod, vm]",
+		"[pod]",
+	} {
+		if _, err := New(task(t, labels), "ddupan.top"); err == nil {
+			t.Fatalf("expected labels %s to fail", labels)
+		}
+	}
+}
+
+func TestAssignmentWireRoundTripAndValidation(t *testing.T) {
+	want, err := New(task(t, "[self-hosted, pod]"), "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	data, err := Marshal(want)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := Unmarshal(data, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.ID != want.ID || got.Backend != want.Backend || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) {
+		t.Fatalf("round trip = %#v, want %#v", got, want)
+	}
+
+	tampered := bytes.Replace(data, []byte(`"backend":"pod"`), []byte(`"backend":"vm"`), 1)
+	if _, err := Unmarshal(tampered, "ddupan.top"); err == nil {
+		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

@@ -0,0 +1,108 @@
+// Package taskidentity derives workload identities from tasks assigned by Gitea.
+package taskidentity
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+
+	"gitea.dev/actionslib/pkg/model"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+)
+
+var safeSegment = regexp.MustCompile(`^[A-Za-z0-9._-]+$`)
+var safeTaskKey = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_-]*$`)
+
+// Identity is the trusted identity context extracted from a fetched task.
+type Identity struct {
+	Repository string
+	Task       string
+	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) {
+	if task == nil || task.Context == nil {
+		return Identity{}, errors.New("task context is required")
+	}
+
+	repository := strings.TrimSpace(task.Context.GetFields()["repository"].GetStringValue())
+	taskName, err := workflowTaskKey(task.WorkflowPayload)
+	if err != nil {
+		return Identity{}, err
+	}
+	parts := strings.Split(repository, "/")
+	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+		return Identity{}, fmt.Errorf("invalid task repository %q", repository)
+	}
+	trustDomain = strings.TrimSpace(trustDomain)
+	if trustDomain == "" || strings.ContainsAny(trustDomain, "/ ") {
+		return Identity{}, fmt.Errorf("invalid trust domain %q", trustDomain)
+	}
+
+	path := strings.Join([]string{
+		sanitize(parts[0]),
+		sanitize(parts[1]),
+		taskName,
+	}, "/")
+	return Identity{
+		Repository: repository,
+		Task:       taskName,
+		SPIFFEID:   "spiffe://" + trustDomain + "/ci/" + path,
+	}, nil
+}
+
+func workflowTaskKey(payload []byte) (string, error) {
+	workflow, err := model.ReadWorkflow(bytes.NewReader(payload))
+	if err != nil {
+		return "", fmt.Errorf("parse task workflow: %w", err)
+	}
+	jobIDs := workflow.GetJobIDs()
+	if len(jobIDs) != 1 {
+		return "", fmt.Errorf("task workflow must contain exactly one job, got %d", len(jobIDs))
+	}
+	if !safeTaskKey.MatchString(jobIDs[0]) {
+		return "", fmt.Errorf("task job key %q must match %s", jobIDs[0], safeTaskKey)
+	}
+	return jobIDs[0], nil
+}
+
+// BackoffTaskSegment deterministically converts a legacy display name into a
+// collision-resistant path segment. Canonical task identities do not use it.
+func BackoffTaskSegment(value string) string {
+	return sanitize(value)
+}
+
+func sanitize(value string) string {
+	if safeSegment.MatchString(value) {
+		return value
+	}
+	slug := strings.Trim(regexp.MustCompile(`[^A-Za-z0-9._-]+`).ReplaceAllString(value, "-"), "-._")
+	if len(slug) > 48 {
+		slug = strings.TrimRight(slug[:48], "-._")
+	}
+	if slug == "" {
+		slug = "segment"
+	}
+	digest := fmt.Sprintf("%x", sha256.Sum256([]byte(value)))[:12]
+	return slug + "-" + digest
+}

internal/taskidentity/identity_test.go

@@ -0,0 +1,77 @@
+package taskidentity
+
+import (
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func TestFromTaskUsesFetchedContext(t *testing.T) {
+	ctx, err := structpb.NewStruct(map[string]any{
+		"repository": "panxiao81/gitea-dynamic-runner",
+		"job":        "Publish images",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := FromTask(&runnerv1.Task{
+		Id:              900,
+		Context:         ctx,
+		WorkflowPayload: []byte("jobs:\n  publish-images:\n    runs-on: [self-hosted, vm]\n    steps: []\n"),
+	}, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Repository != "panxiao81/gitea-dynamic-runner" || got.Task != "publish-images" {
+		t.Fatalf("unexpected task identity context: %#v", got)
+	}
+	want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/publish-images"
+	if got.SPIFFEID != want {
+		t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want)
+	}
+}
+
+func TestFromTaskRejectsUnsafeWorkflowJobKey(t *testing.T) {
+	ctx, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := &runnerv1.Task{
+		Context:         ctx,
+		WorkflowPayload: []byte("jobs:\n  'Run on Ubuntu':\n    runs-on: self-hosted\n    steps: []\n"),
+	}
+	if _, err := FromTask(task, "ddupan.top"); err == nil {
+		t.Fatal("expected unsafe job key to fail")
+	}
+}
+
+func TestBackoffTaskSegmentIsStableAndCollisionResistant(t *testing.T) {
+	got := BackoffTaskSegment("Run on Ubuntu")
+	if got != "Run-on-Ubuntu-8b7cd4c244fb" {
+		t.Fatalf("backoff segment = %q", got)
+	}
+	if got == BackoffTaskSegment("Run@on Ubuntu") {
+		t.Fatal("different legacy names must not collide after slugging")
+	}
+}
+
+func TestFromTaskRejectsIncompleteServerContext(t *testing.T) {
+	for _, fields := range []map[string]any{
+		{"repository": "invalid"},
+		{"repository": ""},
+	} {
+		ctx, err := structpb.NewStruct(fields)
+		if err != nil {
+			t.Fatal(err)
+		}
+		task := &runnerv1.Task{
+			Context:         ctx,
+			WorkflowPayload: []byte("jobs:\n  test:\n    runs-on: self-hosted\n    steps: []\n"),
+		}
+		if _, err := FromTask(task, "ddupan.top"); err == nil {
+			t.Fatalf("expected invalid task context to fail: %#v", fields)
+		}
+	}
+}

internal/taskscheduler/poller.go

@@ -0,0 +1,136 @@
+package taskscheduler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+)
+
+type PollClient interface {
+	Declare(context.Context, string, []string) error
+	FetchTask(context.Context, int64) (*runnerv1.FetchTaskResponse, error)
+}
+
+type PollerConfig struct {
+	Version      string
+	Labels       []string
+	EmptyBackoff time.Duration
+	ErrorBackoff time.Duration
+	Capacity     int
+}
+
+// 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
+	Config    PollerConfig
+	OnError   func(error)
+}
+
+func (p Poller) Run(ctx context.Context) error {
+	if p.Client == nil || p.Scheduler == nil {
+		return errors.New("Gitea poll client and task scheduler are required")
+	}
+	if p.Config.Version == "" || len(p.Config.Labels) == 0 {
+		return errors.New("runner version and labels are required")
+	}
+	if err := p.Client.Declare(ctx, p.Config.Version, p.Config.Labels); err != nil {
+		return fmt.Errorf("declare scheduler labels: %w", err)
+	}
+
+	emptyBackoff := p.Config.EmptyBackoff
+	if emptyBackoff <= 0 {
+		emptyBackoff = time.Second
+	}
+	errorBackoff := p.Config.ErrorBackoff
+	if errorBackoff <= 0 {
+		errorBackoff = 5 * time.Second
+	}
+	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 {
+		response, err := p.Client.FetchTask(ctx, tasksVersion.Load())
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			p.report(fmt.Errorf("fetch Gitea task: %w", err))
+			if !wait(ctx, errorBackoff) {
+				return nil
+			}
+			continue
+		}
+		if response == nil {
+			p.report(errors.New("fetch Gitea task returned an empty response"))
+			if !wait(ctx, errorBackoff) {
+				return nil
+			}
+			continue
+		}
+		storeMaximum(tasksVersion, response.GetTasksVersion())
+		task := response.GetTask()
+		if task == nil {
+			if !wait(ctx, emptyBackoff) {
+				return nil
+			}
+			continue
+		}
+		for {
+			if err := p.Scheduler.Run(ctx, task); err == nil {
+				break
+			} else {
+				if ctx.Err() != nil {
+					return nil
+				}
+				p.report(fmt.Errorf("dispatch Gitea task %d: %w", task.GetId(), err))
+			}
+			if !wait(ctx, errorBackoff) {
+				return nil
+			}
+		}
+	}
+}
+
+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)
+	}
+}
+
+func wait(ctx context.Context, duration time.Duration) bool {
+	timer := time.NewTimer(duration)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
+}

internal/taskscheduler/poller_test.go

@@ -0,0 +1,166 @@
+package taskscheduler
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type fakePollClient struct {
+	mu        sync.Mutex
+	declared  int
+	fetches   int
+	responses []*runnerv1.FetchTaskResponse
+}
+
+func (c *fakePollClient) Declare(context.Context, string, []string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.declared++
+	return nil
+}
+
+func (c *fakePollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) {
+	c.mu.Lock()
+	c.fetches++
+	if len(c.responses) > 0 {
+		response := c.responses[0]
+		c.responses = c.responses[1:]
+		c.mu.Unlock()
+		return response, nil
+	}
+	c.mu.Unlock()
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+type retryDispatcher struct {
+	mu               sync.Mutex
+	calls            int
+	failures         int
+	done             chan struct{}
+	fetches          func() int
+	fetchesAtSuccess int
+}
+
+func (d *retryDispatcher) Dispatch(context.Context, Assignment) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.calls++
+	if d.failures > 0 {
+		d.failures--
+		return errors.New("JetStream unavailable")
+	}
+	if d.fetches != nil {
+		d.fetchesAtSuccess = d.fetches()
+	}
+	select {
+	case <-d.done:
+	default:
+		close(d.done)
+	}
+	return nil
+}
+
+func pollTask(t *testing.T) *runnerv1.Task {
+	t.Helper()
+	fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &runnerv1.Task{
+		Id: 42, Context: fields,
+		WorkflowPayload: []byte("jobs:\n  test:\n    runs-on: [self-hosted, pod]\n    steps: []\n"),
+	}
+}
+
+func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) {
+	client := &fakePollClient{responses: []*runnerv1.FetchTaskResponse{{Task: pollTask(t), TasksVersion: 7}}}
+	dispatcher := &retryDispatcher{failures: 2, done: make(chan struct{})}
+	dispatcher.fetches = func() int {
+		client.mu.Lock()
+		defer client.mu.Unlock()
+		return client.fetches
+	}
+	poller := Poller{
+		Client:    client,
+		Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher},
+		Config: PollerConfig{
+			Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"},
+			EmptyBackoff: time.Millisecond, ErrorBackoff: time.Millisecond,
+		},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	finished := make(chan error, 1)
+	go func() { finished <- poller.Run(ctx) }()
+	select {
+	case <-dispatcher.done:
+	case <-time.After(time.Second):
+		t.Fatal("assignment was not dispatched")
+	}
+	cancel()
+	if err := <-finished; err != nil {
+		t.Fatal(err)
+	}
+	dispatcher.mu.Lock()
+	defer dispatcher.mu.Unlock()
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if dispatcher.calls != 3 || dispatcher.fetchesAtSuccess != 1 || client.declared != 1 {
+		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/taskscheduler/scheduler.go

@@ -0,0 +1,38 @@
+// Package taskscheduler owns tasks fetched through Gitea's RunnerService and
+// dispatches them to an executor only after their identity is known.
+package taskscheduler
+
+import (
+	"context"
+	"errors"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+)
+
+type Assignment = taskassignment.Assignment
+
+// Dispatcher creates exactly one executor for an already assigned Gitea task.
+// It must not register another runner or ask Gitea for a task.
+type Dispatcher interface {
+	Dispatch(context.Context, Assignment) error
+}
+
+// Scheduler implements the TaskRunner boundary used by Gitea Runner's poller.
+type Scheduler struct {
+	TrustDomain string
+	Dispatcher  Dispatcher
+}
+
+// Run derives identity from the fetched task before provisioning its executor.
+func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error {
+	if s.Dispatcher == nil {
+		return errors.New("executor dispatcher is required")
+	}
+	assignment, err := taskassignment.New(task, s.TrustDomain)
+	if err != nil {
+		return err
+	}
+	return s.Dispatcher.Dispatch(ctx, assignment)
+}

internal/taskscheduler/scheduler_test.go

@@ -0,0 +1,56 @@
+package taskscheduler
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type recordingDispatcher struct {
+	assignment Assignment
+}
+
+func (d *recordingDispatcher) Dispatch(_ context.Context, assignment Assignment) error {
+	d.assignment = assignment
+	return nil
+}
+
+func TestRunDerivesIdentityBeforeDispatch(t *testing.T) {
+	taskContext, err := structpb.NewStruct(map[string]any{
+		"repository": "panxiao81/gitea-dynamic-runner",
+		"job":        "test",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := &runnerv1.Task{
+		Id:              42,
+		Context:         taskContext,
+		WorkflowPayload: []byte("jobs:\n  test:\n    runs-on: [self-hosted, pod]\n    steps: []\n"),
+	}
+	dispatcher := &recordingDispatcher{}
+	scheduler := Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher}
+
+	if err := scheduler.Run(context.Background(), task); err != nil {
+		t.Fatal(err)
+	}
+	if dispatcher.assignment.Task != task {
+		t.Fatal("dispatcher did not receive the fetched task")
+	}
+	if dispatcher.assignment.ID != "gitea-task-42" {
+		t.Fatalf("assignment ID = %q", dispatcher.assignment.ID)
+	}
+	want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/test"
+	if dispatcher.assignment.Identity.SPIFFEID != want {
+		t.Fatalf("SPIFFE ID = %q, want %q", dispatcher.assignment.Identity.SPIFFEID, want)
+	}
+}
+
+func TestRunRequiresDispatcher(t *testing.T) {
+	scheduler := Scheduler{TrustDomain: "ddupan.top"}
+	if err := scheduler.Run(context.Background(), &runnerv1.Task{}); err == nil {
+		t.Fatal("expected missing dispatcher to fail")
+	}
+}

internal/taskworker/worker.go

@@ -0,0 +1,220 @@
+// Package taskworker reconciles assigned Gitea tasks against an executor backend.
+package taskworker
+
+import (
+	"context"
+	"errors"
+	"strconv"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+)
+
+// Phase is observed from Kubernetes or OpenSandbox, never stored by the worker.
+type Phase string
+
+const (
+	PhasePending   Phase = "pending"
+	PhaseRunning   Phase = "running"
+	PhaseSucceeded Phase = "succeeded"
+	PhaseFailed    Phase = "failed"
+)
+
+// Executor is the backend resource discovered by assignment ID.
+type Executor struct {
+	Name           string
+	IdentityTarget string
+	Phase          Phase
+}
+
+// Metadata is persisted on the backend resource. Labels remain query-safe;
+// annotations retain complete identity context.
+type Metadata struct {
+	Labels      map[string]string
+	Annotations map[string]string
+}
+
+// LaunchSpec contains the durable resource metadata and the short-lived
+// executor environment. Environment values configure the one-shot runner but
+// are deliberately excluded from labels and annotations.
+type LaunchSpec struct {
+	Metadata    Metadata
+	Environment map[string]string
+}
+
+// Bootstrap produces assignment-scoped executor configuration. Implementations
+// must be deterministic so a redelivery after controller restart creates the
+// same credentials without storing another lifecycle record.
+type Bootstrap interface {
+	Environment(taskassignment.Assignment) (map[string]string, error)
+}
+
+// Backend is implemented by the native Pod and OpenSandbox adapters.
+// Every method must be idempotent.
+type Backend interface {
+	Find(context.Context, string) (*Executor, error)
+	Create(context.Context, taskassignment.Assignment, LaunchSpec) (*Executor, error)
+	BindIdentity(context.Context, *Executor, taskidentity.Identity) error
+	Delete(context.Context, *Executor) error
+}
+
+// TaskState uses Gitea as the authority for whether an assigned task has
+// already reached a terminal state.
+type TaskState interface {
+	Terminal(context.Context, int64) (bool, error)
+	Report(context.Context, int64, Phase) error
+}
+
+// Worker has no correctness-critical in-memory state. Handle may be called
+// again for the same assignment after any operation.
+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
+// returns true, all recovery information exists in Kubernetes/OpenSandbox and
+// the assignment message can be acknowledged immediately.
+func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment) (bool, error) {
+	if w.Backend == nil || w.Bootstrap == nil {
+		return false, errors.New("backend and runner bootstrap are required")
+	}
+	if assignment.ID == "" || assignment.Task == nil {
+		return false, errors.New("valid assignment is required")
+	}
+	executor, err := w.Backend.Find(ctx, assignment.ID)
+	if err != nil {
+		return false, err
+	}
+	if executor == nil {
+		w.event("executor_absent", assignment, nil)
+		launch, launchErr := w.launchSpec(assignment)
+		if launchErr != nil {
+			return false, launchErr
+		}
+		executor, err = w.Backend.Create(ctx, assignment, launch)
+		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
+}
+
+// Handle performs one reconciliation. Done means the queue message may be
+// acknowledged. A false result should remain pending and be reconciled again.
+func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment) (done bool, err error) {
+	if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil {
+		return false, errors.New("backend, Gitea task state, and runner bootstrap are required")
+	}
+	if assignment.ID == "" || assignment.Task == nil {
+		return false, errors.New("valid assignment is required")
+	}
+
+	terminal, err := w.Tasks.Terminal(ctx, assignment.Task.GetId())
+	if err != nil {
+		return false, err
+	}
+	executor, err := w.Backend.Find(ctx, assignment.ID)
+	if err != nil {
+		return false, err
+	}
+	if terminal {
+		if executor != nil {
+			if err := w.Backend.Delete(ctx, executor); err != nil {
+				return false, err
+			}
+		}
+		return true, nil
+	}
+
+	if executor == nil {
+		launch, launchErr := w.launchSpec(assignment)
+		if launchErr != nil {
+			return false, launchErr
+		}
+		executor, err = w.Backend.Create(ctx, assignment, launch)
+		if err != nil {
+			return false, err
+		}
+	}
+	if executor.IdentityTarget == "" {
+		return false, nil
+	}
+	if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
+		return false, err
+	}
+
+	switch executor.Phase {
+	case PhaseSucceeded, PhaseFailed:
+		// Reporting first closes the delete-before-ack crash window: after a
+		// restart Gitea prevents this assignment from executing a second time.
+		if err := w.Tasks.Report(ctx, assignment.Task.GetId(), executor.Phase); err != nil {
+			return false, err
+		}
+		if err := w.Backend.Delete(ctx, executor); err != nil {
+			return false, err
+		}
+		return true, nil
+	default:
+		return false, nil
+	}
+}
+
+func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, error) {
+	environment, err := w.Bootstrap.Environment(assignment)
+	if err != nil {
+		return LaunchSpec{}, err
+	}
+	return LaunchSpec{Metadata: BackendMetadata(assignment), Environment: environment}, nil
+}
+
+// BackendMetadata is the shared metadata contract for Pods and OpenSandbox.
+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),
+		},
+		Annotations: map[string]string{
+			"ci.ddupan.top/repository": assignment.Identity.Repository,
+			"ci.ddupan.top/job-key":    assignment.Identity.Task,
+			"ci.ddupan.top/spiffe-id":  assignment.Identity.SPIFFEID,
+		},
+	}
+}

internal/taskworker/worker_test.go

@@ -0,0 +1,148 @@
+package taskworker
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+)
+
+type fakeBackend struct {
+	executor *Executor
+	created  int
+	bound    int
+	deleted  int
+}
+
+type fakeBootstrap struct{}
+
+func (fakeBootstrap) Environment(taskassignment.Assignment) (map[string]string, error) {
+	return map[string]string{"CI_RUNNER_CAPABILITY": "capability"}, nil
+}
+
+func (b *fakeBackend) Find(context.Context, string) (*Executor, error) { return b.executor, nil }
+func (b *fakeBackend) Create(_ context.Context, _ taskassignment.Assignment, _ LaunchSpec) (*Executor, error) {
+	b.created++
+	b.executor = &Executor{Name: "executor", IdentityTarget: "pod-uid", Phase: PhaseRunning}
+	return b.executor, nil
+}
+func (b *fakeBackend) BindIdentity(context.Context, *Executor, taskidentity.Identity) error {
+	b.bound++
+	return nil
+}
+func (b *fakeBackend) Delete(context.Context, *Executor) error {
+	b.deleted++
+	b.executor = nil
+	return nil
+}
+
+type fakeTasks struct {
+	terminal bool
+	reported []Phase
+}
+
+func (t *fakeTasks) Terminal(context.Context, int64) (bool, error) { return t.terminal, nil }
+func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error {
+	t.reported = append(t.reported, phase)
+	t.terminal = true
+	return nil
+}
+
+func assignment() taskassignment.Assignment {
+	return taskassignment.Assignment{
+		ID:      "gitea-task-42",
+		Backend: taskassignment.BackendPod,
+		Task:    &runnerv1.Task{Id: 42},
+		Identity: taskidentity.Identity{
+			Repository: "owner/repo",
+			Task:       "publish",
+			SPIFFEID:   "spiffe://ddupan.top/ci/owner/repo/publish",
+		},
+	}
+}
+
+func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
+	backend := &fakeBackend{executor: &Executor{Name: "existing", IdentityTarget: "uid", Phase: PhaseRunning}}
+	worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}}
+
+	done, err := worker.Handle(context.Background(), assignment())
+	if err != nil || done {
+		t.Fatalf("Handle() = (%v, %v), want pending", done, err)
+	}
+	if backend.created != 0 || backend.bound != 1 {
+		t.Fatalf("created=%d bound=%d", backend.created, backend.bound)
+	}
+}
+
+func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
+	backend := &fakeBackend{}
+	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 {
+		t.Fatalf("Accept() = (%v, %v), want accepted", accepted, err)
+	}
+	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) {
+	backend := &fakeBackend{executor: &Executor{Name: "pending", Phase: PhasePending}}
+	worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
+
+	accepted, err := worker.Accept(context.Background(), assignment())
+	if err != nil || accepted {
+		t.Fatalf("Accept() = (%v, %v), want retry", accepted, err)
+	}
+	if backend.bound != 0 {
+		t.Fatalf("identity bindings = %d", backend.bound)
+	}
+}
+
+func TestHandleReportsBeforeCleanupAndBecomesRecoverable(t *testing.T) {
+	backend := &fakeBackend{executor: &Executor{Name: "finished", IdentityTarget: "uid", Phase: PhaseSucceeded}}
+	tasks := &fakeTasks{}
+	worker := Worker{Backend: backend, Tasks: tasks, Bootstrap: fakeBootstrap{}}
+
+	done, err := worker.Handle(context.Background(), assignment())
+	if err != nil || !done {
+		t.Fatalf("Handle() = (%v, %v), want done", done, err)
+	}
+	if len(tasks.reported) != 1 || backend.deleted != 1 {
+		t.Fatalf("reported=%v deleted=%d", tasks.reported, backend.deleted)
+	}
+
+	// Simulate redelivery after deletion but before the queue ACK. Gitea's
+	// terminal state prevents a duplicate executor from being created.
+	done, err = worker.Handle(context.Background(), assignment())
+	if err != nil || !done || backend.created != 0 {
+		t.Fatalf("recovery = (%v, %v), created=%d", done, err, backend.created)
+	}
+}
+
+func TestBackendMetadataContainsRecoveryKeys(t *testing.T) {
+	metadata := BackendMetadata(assignment())
+	if metadata.Labels["ci.ddupan.top/assignment-id"] != "gitea-task-42" {
+		t.Fatalf("labels = %#v", metadata.Labels)
+	}
+	if metadata.Annotations["ci.ddupan.top/spiffe-id"] != "spiffe://ddupan.top/ci/owner/repo/publish" {
+		t.Fatalf("annotations = %#v", metadata.Annotations)
+	}
+}