使用 native snapshotter 生成镜像证明

Reviewed-on: #15

.gitea/workflows/pod-smoke.yml

@@ -0,0 +1,19 @@
+name: dynamic Pod smoke test
+
+on:
+  workflow_dispatch:
+
+jobs:
+  pod-smoke:
+    name: pod-smoke
+    runs-on: [self-hosted, pod]
+    timeout-minutes: 5
+    steps:
+      - name: Verify disposable host environment
+        run: |
+          test -S /run/spire/agent-sockets/spire-agent.sock
+          /opt/spire/bin/spire-agent api fetch jwt \
+            -audience ci-smoke \
+            -socketPath /run/spire/agent-sockets/spire-agent.sock \
+            >/dev/null
+          test "$(id -u)" = 0

.gitea/workflows/publish-images.yml

@@ -0,0 +1,154 @@
+---
+name: publish images
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - '.gitea/workflows/publish-images.yml'
+      - 'config/**'
+      - 'container/**'
+      - 'scripts/**'
+      - 'src/**'
+      - 'pyproject.toml'
+      - 'README.md'
+  workflow_dispatch:
+
+jobs:
+  publish-images:
+    name: publish-images
+    runs-on: [self-hosted, pod]
+    timeout-minutes: 45
+    permissions:
+      contents: read
+    env:
+      PUSH_REGISTRY: zot-push.ad.ddupan.top
+      PULL_REGISTRY: zot.ad.ddupan.top
+      CONTROLLER_REPOSITORY: panxiao81/gitea-dynamic-runner-controller
+      RUNNER_REPOSITORY: panxiao81/gitea-dynamic-runner-runner
+      SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Test source
+        shell: bash
+        run: |
+          set -euo pipefail
+          python3 -m pip install --break-system-packages -e '.[test]'
+          pytest -q
+          python3 -m compileall -q src tests
+          apt-get update
+          apt-get install --yes --no-install-recommends shellcheck
+          shellcheck scripts/*
+
+      - name: Start Docker
+        shell: bash
+        run: |
+          set -euo pipefail
+          # The runner Pod itself uses an overlay filesystem. A nested overlay
+          # snapshotter cannot mount there, so use the copy-based vfs driver.
+          dockerd \
+            --storage-driver=vfs \
+            --feature containerd-snapshotter=false \
+            >/tmp/dockerd.log 2>&1 &
+          for _ in $(seq 1 60); do
+            if docker info >/dev/null 2>&1; then
+              exit 0
+            fi
+            if ! kill -0 "$!" 2>/dev/null; then
+              cat /tmp/dockerd.log >&2
+              exit 1
+            fi
+            sleep 1
+          done
+          cat /tmp/dockerd.log >&2
+          exit 1
+
+      - name: Build and publish
+        shell: bash
+        run: |
+          set -euo pipefail
+          set +x
+
+          : "${GITHUB_SHA:?GITHUB_SHA is required}"
+          image_tag="sha-${GITHUB_SHA}"
+          docker_config=$(mktemp -d)
+          jwt_file=$(mktemp)
+          buildkit_config=$(mktemp)
+          cleanup() {
+            docker buildx rm ci-builder >/dev/null 2>&1 || true
+            rm -rf -- "$docker_config" "$jwt_file" "$buildkit_config"
+          }
+          trap cleanup EXIT
+          export DOCKER_CONFIG="$docker_config"
+
+          /opt/spire/bin/spire-agent api fetch jwt \
+            -audience zot \
+            -socketPath "$SPIRE_AGENT_SOCKET" \
+            -output json >"$jwt_file"
+          # shellcheck disable=SC2016
+          jq -er '.[0].svids[0].svid' "$jwt_file" | \
+            docker login "$PUSH_REGISTRY" --username zot --password-stdin
+
+          # The docker driver cannot publish attestations. Use an isolated
+          # BuildKit worker, but force its snapshotter to copy-based native so
+          # it does not attempt nested overlay mounts inside the runner Pod.
+          printf '%s\n' \
+            '[worker.oci]' \
+            '  snapshotter = "native"' \
+            >"$buildkit_config"
+          docker buildx create \
+            --name ci-builder \
+            --driver docker-container \
+            --buildkitd-config "$buildkit_config" \
+            --use
+
+          publish() {
+            local repository=$1
+            local dockerfile=$2
+            local metadata=$3
+            docker buildx build \
+              --builder ci-builder \
+              --platform linux/amd64 \
+              --file "$dockerfile" \
+              --tag "${PUSH_REGISTRY}/${repository}:${image_tag}" \
+              --tag "${PUSH_REGISTRY}/${repository}:main" \
+              --provenance=mode=max \
+              --sbom=true \
+              --metadata-file "$metadata" \
+              --push \
+              .
+          }
+
+          publish \
+            "$CONTROLLER_REPOSITORY" \
+            container/controller.Dockerfile \
+            controller-metadata.json
+          publish \
+            "$RUNNER_REPOSITORY" \
+            container/runner.Dockerfile \
+            runner-metadata.json
+
+          controller_digest=$(
+            # shellcheck disable=SC2016
+            jq -er '."containerimage.digest"' controller-metadata.json
+          )
+          runner_digest=$(
+            # shellcheck disable=SC2016
+            jq -er '."containerimage.digest"' runner-metadata.json
+          )
+          controller_ref="${PULL_REGISTRY}/${CONTROLLER_REPOSITORY}@${controller_digest}"
+          runner_ref="${PULL_REGISTRY}/${RUNNER_REPOSITORY}@${runner_digest}"
+
+          printf 'controller=%s\nrunner=%s\n' "$controller_ref" "$runner_ref"
+          if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
+            {
+              printf '## Published images\n\n'
+              # shellcheck disable=SC2016
+              printf -- '- Controller: `%s`\n' "$controller_ref"
+              # shellcheck disable=SC2016
+              printf -- '- Runner: `%s`\n' "$runner_ref"
+              # shellcheck disable=SC2016
+              printf -- '- Source: `%s`\n' "$GITHUB_SHA"
+            } >>"$GITHUB_STEP_SUMMARY"
+          fi

.gitea/workflows/spire-identity-smoke.yml

@@ -0,0 +1,33 @@
+name: SPIRE identity smoke
+
+on:
+  workflow_dispatch:
+
+jobs:
+  jwt-svid:
+    runs-on: self-hosted
+    container:
+      volumes:
+        - /run/spire/agent-sockets:/run/spire/agent-sockets:ro
+    steps:
+      - name: Fetch pinned SPIRE CLI
+        shell: bash
+        run: |
+          set -euo pipefail
+          archive=/tmp/spire.tar.gz
+          curl --fail --location --silent --show-error \
+            --output "$archive" \
+            https://github.com/spiffe/spire/releases/download/v1.15.3/spire-1.15.3-linux-amd64-musl.tar.gz
+          printf '%s  %s\n' \
+            ca1a4d1155317bdd2afc7f36663828a10410c7c840e54725b90b4064b0a301c7 \
+            "$archive" | sha256sum --check --status
+          tar -xzf "$archive" -C /tmp spire-1.15.3/bin/spire-agent
+
+      - name: Fetch short-lived zot JWT-SVID
+        shell: bash
+        run: |
+          set -euo pipefail
+          /tmp/spire-1.15.3/bin/spire-agent api fetch jwt \
+            -audience zot \
+            -socketPath /run/spire/agent-sockets/spire-agent.sock \
+            >/dev/null

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

@@ -0,0 +1,31 @@
+name: VM kind smoke
+
+on:
+  workflow_dispatch:
+
+jobs:
+  kind:
+    runs-on: [self-hosted, vm]
+    steps:
+      - name: Verify Docker
+        run: docker info
+
+      - name: Install kind
+        shell: bash
+        run: |
+          set -euo pipefail
+          version=v0.33.0
+          curl --fail --location --silent --show-error \
+            --output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/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
+          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
+          /tmp/kind get clusters | grep -Fx smoke

.github/actionlint.yaml

@@ -0,0 +1,5 @@
+---
+self-hosted-runner:
+  labels:
+    - pod
+    - vm

README.md

@@ -1,21 +1,43 @@
-# Gitea microVM runner
+# Gitea dynamic runner
 
-为 Gitea Actions 按需启动 Cloud Hypervisor microVM。适合 kind、嵌套容器和其他不应
+为 Gitea Actions 按需创建一次性执行环境。对 workflow 提供两种稳定的 runner
-在常驻 Kubernetes runner 中执行的 CI 工作负载。
+接口：
+
+```yaml
+runs-on: [self-hosted, pod]
+```
+
+```yaml
+runs-on: [self-hosted, vm]
+```
+
+`pod` 使用动态 Kubernetes Pod，`vm` 使用动态 Cloud Hypervisor microVM。每个环境
+只执行一个 job，并在 job 结束后连同本地状态一起销毁。完整的设计约束见
+[`docs/design-principles.md`](docs/design-principles.md)。
 
 组件：
 
 - `controller`：接收 Gitea `workflow_job` webhook，将指定 label 的 queued job
   发布到 NATS JetStream。
-- `worker`：在虚拟化宿主机领取任务，限制本机并发，并启动一次性 microVM。
+- `worker`：领取任务、限制并发，并通过 Pod 或 microVM backend 创建一次性环境。
-- `microvm-runner-launch`：为每个任务创建 COW disk、NoCloud seed 和 TAP，运行
+- `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行
   Cloud Hypervisor，退出后完整清理。
 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral
   runner，执行一个 job 后关机。
+- `pod-worker`：在 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 身份。
 
 消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一
 durable consumer；扩容只需要增加 worker 或提高单机 capacity。
 
+当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现，不是
+长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
+声明 labels、领取 task，并把已领取 task 交给 Pod/VM executor；路线与迁移边界见
+[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
+
 ## 开发
 
 ```bash
@@ -29,9 +51,16 @@ pytest
 
 - NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。
 - registration token 不写入 seed image；worker 通过单次 nonce endpoint 交给 guest。
+- guest 启动时从仅监听 microVM bridge 的 worker endpoint 获取固定版本 Runner 和配置
+  资产；基础镜像无需为 Runner 发布而重做。
+- `runner-vm-bootstrap.yaml` 暂时只验证 VM 调度和生命周期，不提供 SPIFFE
+  identity；VM agent attestation 完成前不得将它当作身份链路验证结果。
 - guest runner 使用 `--ephemeral`，每台 VM 只执行一个 job。
 - launcher 只接受 UUID instance ID 和 URL-safe nonce，所有临时文件都位于独立目录。
 - base image 不得包含 runner identity、registration token、SSH 密码或 host key。
+- LXC 内运行 Cloud Hypervisor 必须为 guest memory 启用 `shared=on`。默认的 private
+  memfd 映射会在 guest 写入后同时产生 shmem 与 anonymous CoW charge，使 LXC cgroup
+  对 guest RAM 接近双倍计费。
 
 homelab 的 Kubernetes、OpenBao、LXC、bridge 和容量配置保留在
 `panxiao81/homelab-infra`。

config/runner-vm-bootstrap.yaml

@@ -0,0 +1,11 @@
+runner:
+  capacity: 1
+  timeout: 3h
+  shutdown_timeout: 1m
+
+host:
+  workdir_parent: /workspace
+
+container:
+  require_docker: false
+  valid_volumes: []

config/runner.yaml

@@ -0,0 +1,13 @@
+runner:
+  capacity: 1
+  timeout: 3h
+  shutdown_timeout: 1m
+  hooks:
+    job_started: /usr/local/libexec/gitea-job-started
+
+host:
+  workdir_parent: /workspace
+
+container:
+  require_docker: false
+  valid_volumes: []

container/controller.Dockerfile

@@ -1,3 +1,5 @@
+FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
+
 FROM python:3.12.11-alpine3.22 AS build
 WORKDIR /src
 COPY pyproject.toml README.md ./
@@ -5,8 +7,11 @@ COPY src ./src
 RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir .
 
 FROM python:3.12.11-alpine3.22
-RUN addgroup -S -g 65532 runner && adduser -S -D -H -u 65532 -G runner runner
+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
 USER 65532:65532
 EXPOSE 8787
-ENTRYPOINT ["/venv/bin/gitea-microvm-controller"]
+ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"]

container/runner.Dockerfile

@@ -0,0 +1,20 @@
+FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
+
+FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS 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
+# actions share a GitHub Actions-compatible userspace.
+FROM docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
+USER root
+
+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=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
+
+VOLUME ["/data"]
+WORKDIR /
+ENTRYPOINT ["/usr/local/bin/run.sh"]

docs/design-principles.md

@@ -0,0 +1,88 @@
+# 动态 Runner 设计原则
+
+## 对 workflow 的接口
+
+Runner 只向 workflow 暴露两个执行环境：
+
+```yaml
+runs-on: [self-hosted, pod]
+```
+
+```yaml
+runs-on: [self-hosted, vm]
+```
+
+- `self-hosted` 是固定前缀。
+- `pod` 表示一次性 Kubernetes Pod，承担常规 CI、镜像构建和 kind 等任务。
+- `vm` 表示一次性 microVM，承担需要独立内核、KVM、systemd 或更强隔离的任务。
+
+执行后端是基础设施选择，不是权限角色。workflow 不需要额外声明由 controller
+维护的 role 或权限 label。
+
+## 一个 job，一个环境
+
+Controller 根据 Gitea `workflow_job` webhook 创建执行环境。每个 Pod 或 VM 注册一个
+ephemeral runner，只执行一个 job；任务结束后注销 runner，并删除计算环境及其全部
+本地状态。
+
+`job_id` 仅用于消息去重、状态追踪、实例关联和失败清理，不进入 workload 身份，也
+不参与资源授权。
+
+Gitea 不保证由某次 `queued` webhook 创建的 runner 一定领取该 webhook 对应的 job。
+因此创建环境时只赋予无业务权限的启动身份。runner 实际领取任务后，controller 根据
+`in_progress` webhook 返回的 `runner_name` 和真实 job 名称绑定业务身份；环境中的
+job-start hook 必须等目标 SVID 可用后才放行 workflow 的第一步。不能依据 queued
+事件提前赋予任务权限。
+
+## 环境只提供运行边界
+
+基础镜像只提供启动 runner 和执行 workflow 所需的最小环境。Docker、BuildKit、
+kind 等工具由 pipeline 按需安装和启动，而不是由 controller 预制成常驻服务。
+
+例如 Pod job 可以在 Pod 内启动仅供本次任务使用的 Docker daemon。该 daemon 及其
+镜像、容器和缓存属于当前 job 的临时状态，随 Pod 一起销毁。Docker 创建的容器不是
+独立的身份边界；需要访问凭据的操作由 Pod 中的 workflow 进程完成，并通过环境变量
+或标准输入把短期凭据交给具体工具。
+
+## Workload 身份
+
+动态 Pod 和 VM 都直接拥有自己的 SPIFFE 身份，不继承常驻 runner 的共享身份：
+
+- Pod 通过 Kubernetes workload attestation 取得身份。
+- VM 通过 VM 内的 SPIRE Agent 取得身份。
+
+SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生：
+
+```text
+spiffe://ddupan.top/ci/<owner>/<repository>/<job-name>
+```
+
+同一种任务在不同运行中使用相同的逻辑 SPIFFE ID；每次运行取得独立、短期的 SVID。
+Pod 与 VM 是可替换的执行实现，因此默认不写入 SPIFFE ID。
+
+job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在发生冲突时
+拒绝创建环境，不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用
+不同的 job 名称；workflow 文件只是编排载体，不进入权限身份。
+
+## Self-service 与授权边界
+
+新增 workflow 或 job 时，controller 自动为它派生身份，不维护第二份任务或角色
+allowlist。能够修改仓库 CI 的主体本来就能修改该仓库已有任务，因此 controller 的
+重复审批不能形成额外的安全边界，只会破坏 self-service。
+
+身份不等于权限。新任务可以立即取得自己的 SPIFFE ID，但默认不会因此获得 Zot、
+OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 job 身份
+配置授权策略。
+
+## 非目标设计
+
+目标架构不依赖以下机制：
+
+- 多个 job 共享的常驻 Docker daemon。
+- 常驻 runner Pod 的共享 SPIFFE 身份。
+- 为嵌套 CI 容器转发共享身份的 JWT broker。
+- 将 Gitea 数字 job ID 编入 SPIFFE ID。
+- controller 维护的仓库任务权限 allowlist。
+
+仓库中的 `jwt-broker` 是早期方案的实验实现，在 Pod/VM 动态执行环境完成迁移后不应
+部署。

docs/runner-protocol-roadmap.md

@@ -0,0 +1,59 @@
+# Gitea Runner 协议调度器路线
+
+## 目标
+
+长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
+协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；它只在后端存在可用容量
+时领取 task，然后将该 task 交给一个一次性 Pod 或 microVM 执行。
+
+```text
+Gitea RunnerService
+        │ Register / Declare / FetchTask
+        ▼
+dynamic-runner scheduler
+        │ 已领取的 task + lease
+        ├── Pod executor
+        └── microVM executor
+                │ logs / state / result
+                └──────────────────────► Gitea
+```
+
+这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配，
+不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor
+执行已经领取的 task，并继续完成日志、状态、心跳、取消和最终结果上报。
+
+## 设计约束
+
+- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
+- scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。
+- 每个 executor 只执行一个 task，完成后销毁。
+- SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生，不需要 queued 与
+  in-progress webhook 的二阶段关联。
+- scheduler 的 runner registration credential 不进入 executor；executor 只得到执行
+  当前 task 所需的短期 lease/capability。
+- task ACK、心跳和结果必须能够跨 scheduler 重启恢复；NATS 可以继续作为内部 handoff，
+  但不是 Gitea 任务事实来源。
+- Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。
+
+## 实现顺序
+
+1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本，记录兼容
+   范围并建立协议契约测试。
+2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike，暂不执行
+   task。
+3. 从 act_runner 提取或复用 task 执行与日志上报能力，定义 scheduler 到 executor 的
+   单任务协议。
+4. 首先接入 Pod executor，验证成功、失败、取消、超时和 scheduler 重启。
+5. 接入 microVM executor，并复用同一 task 协议和身份派生逻辑。
+6. 双轨运行并验证后，移除 webhook receiver、临时 runner 注册和 identity binding
+   subject。
+
+## Bootstrap 实现的退出条件
+
+只有同时满足以下条件才能删除 webhook 路径：
+
+- scheduler 能通过 RunnerService 稳定领取并执行 Pod/VM task；
+- Gitea UI 中的实时日志、取消、超时和结论与官方 runner 行为一致；
+- scheduler 重启不会丢失已领取 task，也不会重复执行；
+- SPIFFE 身份只来自实际领取 task；
+- 同一套 workflow 无需修改 `runs-on` 即可从 bootstrap 迁移。

pyproject.toml

@@ -3,9 +3,9 @@ requires = ["setuptools>=75"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "gitea-microvm-runner"
+name = "gitea-dynamic-runner"
-version = "0.1.0"
+version = "0.3.0"
-description = "On-demand Cloud Hypervisor runners for Gitea Actions"
+description = "On-demand Pod and microVM execution environments for Gitea Actions"
 requires-python = ">=3.11"
 dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
 
@@ -13,8 +13,10 @@ dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
 test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
 
 [project.scripts]
-gitea-microvm-controller = "gitea_microvm_runner.controller:main"
+gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main"
-gitea-microvm-worker = "gitea_microvm_runner.worker:cli"
+gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli"
+gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli"
+gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

scripts/gitea-job-started

@@ -0,0 +1,19 @@
+#!/bin/sh
+set -eu
+
+socket=${SPIRE_AGENT_SOCKET:-/run/spire/agent-sockets/spire-agent.sock}
+wait_seconds=${SPIFFE_IDENTITY_WAIT_SECONDS:-60}
+deadline=$(( $(date +%s) + wait_seconds ))
+
+while [ "$(date +%s)" -lt "$deadline" ]; do
+    if timeout 2 /opt/spire/bin/spire-agent api fetch jwt \
+        -audience ci-job-ready \
+        -socketPath "$socket" \
+        >/dev/null 2>&1; then
+        exit 0
+    fi
+    sleep 1
+done
+
+echo "timed out waiting for the job SPIFFE identity" >&2
+exit 1

scripts/gitea-microvm-guest-runner

@@ -4,8 +4,10 @@ set -eu
 token_url=${1:?token URL is required}
 instance=${2:?Gitea instance is required}
 runner_name=${3:?runner name is required}
-runner_label=${4:?runner label is required}
+runner_labels=${4:?runner labels are required}
 token_file=/run/gitea-runner-registration-token
+config_file=${GITEA_RUNNER_CONFIG_FILE:-/etc/gitea-runner/config-vm-bootstrap.yaml}
+export HOME="${HOME:-/root}"
 
 cleanup() {
     rm -f -- "$token_file"
@@ -22,7 +24,8 @@ gitea-runner register \
     --ephemeral \
     --instance "$instance" \
     --name "$runner_name" \
-    --labels "$runner_label:host" \
+    --labels "$runner_labels" \
     --token-file "$token_file"
 rm -f -- "$token_file"
-gitea-runner daemon
+gitea-runner daemon --config "$config_file" --once
+echo gitea-runner-job-complete >/dev/ttyS0

scripts/microvm-runner-launch

@@ -21,7 +21,7 @@ vm_timeout=${RUNNER_VM_TIMEOUT:-3h}
 cpus=${RUNNER_VM_CPUS:-4}
 memory=${RUNNER_VM_MEMORY:-3G}
 gitea_instance=${GITEA_INSTANCE:-https://git.ddupan.top}
-runner_label=${RUNNER_LABEL:-kind-microvm}
+runner_labels=${RUNNER_LABELS:-self-hosted:host,vm:host}
 token_url=${RUNNER_TOKEN_URL:-http://172.30.0.1:8787}
 
 vm_dir="$state_root/instances/$instance_id"
@@ -29,8 +29,13 @@ tap="mvr${instance_id%%-*}"
 overlay="$vm_dir/root.qcow2"
 seed="$vm_dir/seed.img"
 serial="$vm_dir/serial.log"
+log_dir="$state_root/logs"
 
 cleanup() {
+    if test -f "$serial"; then
+        install -d -m 0700 "$log_dir"
+        cp "$serial" "$log_dir/$instance_id.log"
+    fi
     ip link delete "$tap" 2>/dev/null || true
     rm -rf -- "$vm_dir"
 }
@@ -39,7 +44,10 @@ trap cleanup EXIT INT TERM
 test -r "$base_image"
 test -r "$firmware"
 install -d -m 0700 "$state_root/instances" "$vm_dir"
-qemu-img create -q -f qcow2 -F qcow2 -b "$base_image" "$overlay"
+# Cloud Hypervisor cannot open qcow2 backing chains.  Keep each disposable
+# root disk flat and bypass the LXC page cache: caching both a 5 GiB copy and
+# guest RAM can otherwise trigger the container memory limit.
+qemu-img convert -q -T none -t none -f qcow2 -O qcow2 "$base_image" "$overlay"
 
 cat >"$vm_dir/meta-data" <<EOF
 instance-id: $instance_id
@@ -48,7 +56,8 @@ EOF
 cat >"$vm_dir/user-data" <<EOF
 #cloud-config
 runcmd:
-  - [ /usr/local/libexec/gitea-microvm-guest-runner, "$token_url/token/$nonce", "$gitea_instance", "gitea-${instance_id%%-*}", "$runner_label" ]
+  - [ sh, -c, "curl --fail --silent --show-error --retry 10 --retry-all-errors $token_url/assets/guest-assets.tar.gz | tar -xz -C /" ]
+  - [ /usr/local/libexec/gitea-microvm-guest-runner, "$token_url/token/$nonce", "$gitea_instance", "gitea-vm-${instance_id%%-*}", "$runner_labels" ]
 EOF
 cloud-localds "$seed" "$vm_dir/user-data" "$vm_dir/meta-data"
 
@@ -63,12 +72,16 @@ ip tuntap add dev "$tap" mode tap
 ip link set "$tap" master "$bridge"
 ip link set "$tap" up
 
-timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
+if timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
-    --firmware "$firmware" \
+        --firmware "$firmware" \
-    --cpus "boot=$cpus" \
+        --cpus "boot=$cpus" \
-    --memory "size=$memory" \
+        --memory "size=$memory,shared=on" \
-    --disk "path=$overlay" \
+        --disk "path=$overlay,image_type=qcow2,direct=on,sparse=off" \
-    --disk "path=$seed,readonly=on" \
+        --disk "path=$seed,readonly=on,image_type=raw,direct=on,sparse=off" \
-    --net "tap=$tap,mac=$mac" \
+        --net "tap=$tap,mac=$mac" \
-    --serial "file=$serial" \
+        --serial "file=$serial" \
-    --console off
+        --console off; then
+    grep -Fq gitea-runner-job-complete "$serial"
+else
+    exit $?
+fi

src/gitea_dynamic_runner/__init__.py

@@ -0,0 +1 @@
+"""Dynamic Pod and microVM execution environments for Gitea Actions."""

src/gitea_dynamic_runner/controller.py

@@ -4,6 +4,7 @@
 import hashlib
 import hmac
 import json
+import logging
 import os
 import ssl
 from pathlib import Path
@@ -13,8 +14,11 @@ from aiohttp import web
 from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
 from nats.js.errors import NotFoundError
 
-LABEL = os.environ.get("RUNNER_LABEL", "kind-microvm")
+from .models import IdentityBinding, RunnerRequest
-SUBJECT = os.environ.get("NATS_SUBJECT", f"ci.runner.{LABEL}")
+
+
+LOG = logging.getLogger(__name__)
+SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
 STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
 NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
 NATS_USER = os.environ.get("NATS_USER", "ci-producer")
@@ -24,15 +28,9 @@ WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/g
 
 
 def accepts(payload: object) -> tuple[bool, str | None]:
-    if not isinstance(payload, dict) or payload.get("action") != "queued":
+    """Compatibility helper for callers that only need acceptance and identity."""
-        return False, None
+    request = RunnerRequest.from_webhook(payload)
-    job = payload.get("workflow_job")
+    return (request is not None, str(request.job_id) if request else None)
-    if not isinstance(job, dict) or LABEL not in job.get("labels", []):
-        return False, None
-    job_id = job.get("id")
-    if not isinstance(job_id, int) or isinstance(job_id, bool):
-        return False, None
-    return True, str(job_id)
 
 
 def valid_signature(body: bytes, signature: str) -> bool:
@@ -43,7 +41,7 @@ def valid_signature(body: bytes, signature: str) -> bool:
 async def ensure_stream(js: object) -> None:
     config = StreamConfig(
         name=STREAM,
-        subjects=["ci.runner.*"],
+        subjects=[f"{SUBJECT_PREFIX}.>"],
         retention=RetentionPolicy.WORK_QUEUE,
         storage=StorageType.FILE,
         discard=DiscardPolicy.OLD,
@@ -68,15 +66,68 @@ async def webhook(request: web.Request) -> web.Response:
         payload = json.loads(body)
     except json.JSONDecodeError as error:
         raise web.HTTPBadRequest(text="invalid JSON\n") from error
-    accepted, job_id = accepts(payload)
+    context = _webhook_context(payload)
-    if not accepted:
+    LOG.info("workflow_job webhook received %s", context)
-        return web.Response(status=204)
+
-    await request.app["js"].publish(
+    runner_request = RunnerRequest.from_webhook(payload)
-        SUBJECT,
+    if runner_request is not None:
-        body,
+        subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
-        headers={"Nats-Msg-Id": f"gitea-workflow-job-{job_id}"},
+        await request.app["js"].publish(
+            subject,
+            runner_request.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
+        )
+        LOG.info(
+            "runner request published subject=%s job_id=%d run_id=%d backend=%s "
+            "repository=%s job_name=%r labels=%s",
+            subject,
+            runner_request.job_id,
+            runner_request.run_id,
+            runner_request.backend,
+            runner_request.repository,
+            runner_request.job_name,
+            runner_request.labels,
+        )
+        return web.Response(status=202, text="queued\n")
+
+    binding = IdentityBinding.from_webhook(payload)
+    if binding is not None:
+        subject = f"{SUBJECT_PREFIX}.{binding.backend}.binding"
+        await request.app["js"].publish(
+            subject,
+            binding.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
+        )
+        LOG.info(
+            "identity binding published subject=%s job_id=%d run_id=%d backend=%s "
+            "runner_name=%s repository=%s job_name=%r",
+            subject,
+            binding.job_id,
+            binding.run_id,
+            binding.backend,
+            binding.runner_name,
+            binding.repository,
+            binding.job_name,
+        )
+        return web.Response(status=202, text="binding queued\n")
+
+    LOG.warning("workflow_job webhook ignored %s", context)
+    return web.Response(status=204)
+
+
+def _webhook_context(payload: object) -> str:
+    if not isinstance(payload, dict):
+        return f"payload_type={type(payload).__name__}"
+    job = payload.get("workflow_job")
+    repository = payload.get("repository")
+    job = job if isinstance(job, dict) else {}
+    repository = repository if isinstance(repository, dict) else {}
+    return (
+        f"action={payload.get('action')!r} job_id={job.get('id')!r} "
+        f"run_id={job.get('run_id')!r} runner_name={job.get('runner_name')!r} "
+        f"repository={repository.get('full_name')!r} job_name={job.get('name')!r} "
+        f"labels={job.get('labels')!r}"
     )
-    return web.Response(status=202, text="queued\n")
 
 
 async def health(request: web.Request) -> web.Response:
@@ -109,6 +160,7 @@ def create_app() -> web.Application:
 
 
 def main() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
     web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
 
 

src/gitea_dynamic_runner/jwt_broker.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""Fixed-audience JWT-SVID broker for nested Gitea job containers."""
+
+import asyncio
+import json
+import os
+
+from aiohttp import web
+
+SPIRE_AGENT = os.environ.get("SPIRE_AGENT", "/opt/spire/bin/spire-agent")
+SPIRE_SOCKET = os.environ.get(
+    "SPIRE_SOCKET", "/run/spire/agent-sockets/spire-agent.sock"
+)
+AUDIENCE = os.environ.get("JWT_AUDIENCE", "zot")
+MAX_CONCURRENCY = int(os.environ.get("MAX_CONCURRENCY", "4"))
+semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
+
+
+def extract_svid(document: object) -> str:
+    if not isinstance(document, list):
+        raise ValueError("unexpected SPIRE response")
+    for item in document:
+        if not isinstance(item, dict):
+            continue
+        svids = item.get("svids")
+        if not isinstance(svids, list):
+            continue
+        for svid in svids:
+            if isinstance(svid, dict) and isinstance(svid.get("svid"), str):
+                return svid["svid"]
+    raise ValueError("SPIRE response contains no JWT-SVID")
+
+
+async def fetch_svid() -> str:
+    async with semaphore:
+        process = await asyncio.create_subprocess_exec(
+            SPIRE_AGENT,
+            "api",
+            "fetch",
+            "jwt",
+            "-output",
+            "json",
+            "-audience",
+            AUDIENCE,
+            "-socketPath",
+            SPIRE_SOCKET,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=10)
+    if process.returncode != 0:
+        message = stderr.decode(errors="replace").strip()
+        raise RuntimeError(f"SPIRE agent failed: {message}")
+    return extract_svid(json.loads(stdout))
+
+
+async def token(_: web.Request) -> web.Response:
+    try:
+        value = await fetch_svid()
+    except (RuntimeError, ValueError, json.JSONDecodeError, asyncio.TimeoutError):
+        raise web.HTTPServiceUnavailable(text="identity unavailable\n")
+    return web.Response(
+        text=f"{value}\n",
+        content_type="text/plain",
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+async def health(_: web.Request) -> web.Response:
+    return web.Response(text="ok\n")
+
+
+def create_app() -> web.Application:
+    app = web.Application(client_max_size=1024)
+    app.router.add_post("/token", token)
+    app.router.add_get("/healthz", health)
+    return app
+
+
+def main() -> None:
+    web.run_app(
+        create_app(),
+        host=os.environ.get("LISTEN", "0.0.0.0"),
+        port=int(os.environ.get("PORT", "8788")),
+        access_log=None,
+    )
+
+
+if __name__ == "__main__":
+    main()

src/gitea_dynamic_runner/kubernetes.py

@@ -0,0 +1,89 @@
+"""Small in-cluster Kubernetes API client used by the Pod backend."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from urllib.parse import quote
+
+from aiohttp import ClientResponseError, ClientSession, TCPConnector
+import ssl
+
+
+class KubernetesClient:
+    def __init__(
+        self,
+        *,
+        api_url: str,
+        token_file: Path,
+        ca_file: Path,
+        namespace: str,
+    ) -> None:
+        self.api_url = api_url.rstrip("/")
+        self.token_file = token_file
+        self.ca_file = ca_file
+        self.namespace = namespace
+        self.session: ClientSession | None = None
+
+    async def __aenter__(self) -> KubernetesClient:
+        context = ssl.create_default_context(cafile=self.ca_file)
+        self.session = ClientSession(
+            connector=TCPConnector(ssl=context),
+            headers={
+                "Authorization": f"Bearer {self.token_file.read_text().strip()}",
+            },
+            raise_for_status=True,
+        )
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        if self.session is not None:
+            await self.session.close()
+
+    async def create_pod(self, manifest: dict[str, object]) -> dict[str, object]:
+        response = await self._request("POST", self._pods_path(), json=manifest)
+        return await response.json()
+
+    async def get_pod(self, name: str) -> dict[str, object] | None:
+        try:
+            response = await self._request("GET", f"{self._pods_path()}/{quote(name)}")
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def bind_identity(self, name: str, identity_path: str) -> None:
+        patch = {
+            "metadata": {
+                "labels": {"ci.ddupan.top/identity-bound": "true"},
+                "annotations": {"ci.ddupan.top/spiffe-path": identity_path},
+            }
+        }
+        response = await self._request(
+            "PATCH",
+            f"{self._pods_path()}/{quote(name)}",
+            data=json.dumps(patch),
+            headers={"Content-Type": "application/merge-patch+json"},
+        )
+        response.release()
+
+    async def delete_pod(self, name: str) -> None:
+        try:
+            response = await self._request(
+                "DELETE",
+                f"{self._pods_path()}/{quote(name)}",
+                json={"gracePeriodSeconds": 30, "propagationPolicy": "Background"},
+            )
+            response.release()
+        except ClientResponseError as error:
+            if error.status != 404:
+                raise
+
+    async def _request(self, method: str, path: str, **kwargs: object):
+        if self.session is None:
+            raise RuntimeError("KubernetesClient is not open")
+        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
+
+    def _pods_path(self) -> str:
+        return f"/api/v1/namespaces/{quote(self.namespace)}/pods"

src/gitea_dynamic_runner/models.py

@@ -0,0 +1,176 @@
+"""Validated messages shared by the controller and runner backends."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import json
+
+
+BACKENDS = frozenset({"pod", "vm"})
+REQUIRED_LABEL = "self-hosted"
+
+
+@dataclass(frozen=True, slots=True)
+class RunnerRequest:
+    """A queued Gitea job that needs one disposable runner environment."""
+
+    job_id: int
+    run_id: int
+    backend: str
+    repository: str
+    job_name: str
+    labels: tuple[str, ...]
+    schema_version: int = 1
+
+    @classmethod
+    def from_webhook(cls, payload: object) -> RunnerRequest | None:
+        if not isinstance(payload, dict) or payload.get("action") != "queued":
+            return None
+
+        job = payload.get("workflow_job")
+        repository = payload.get("repository")
+        if not isinstance(job, dict) or not isinstance(repository, dict):
+            return None
+
+        labels_value = job.get("labels")
+        if not isinstance(labels_value, list) or not all(
+            isinstance(label, str) for label in labels_value
+        ):
+            return None
+        labels = tuple(dict.fromkeys(labels_value))
+        selected = BACKENDS.intersection(labels)
+        if REQUIRED_LABEL not in labels or len(selected) != 1:
+            return None
+
+        job_id = job.get("id")
+        run_id = job.get("run_id")
+        job_name = job.get("name")
+        full_name = repository.get("full_name")
+        if not _positive_int(job_id) or not _positive_int(run_id):
+            return None
+        if not all(_nonempty(value) for value in (job_name, full_name)):
+            return None
+
+        return cls(
+            job_id=job_id,
+            run_id=run_id,
+            backend=next(iter(selected)),
+            repository=full_name.strip(),
+            job_name=job_name.strip(),
+            labels=labels,
+        )
+
+    def to_json(self) -> bytes:
+        document = asdict(self)
+        document["labels"] = list(self.labels)
+        return json.dumps(
+            document,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode()
+
+    @classmethod
+    def from_json(cls, body: bytes) -> RunnerRequest:
+        document = json.loads(body)
+        if not isinstance(document, dict) or document.get("schema_version") != 1:
+            raise ValueError("unsupported runner request")
+        labels = document.get("labels")
+        if not isinstance(labels, list) or not all(
+            isinstance(label, str) for label in labels
+        ):
+            raise ValueError("invalid runner labels")
+        try:
+            request = cls(
+                job_id=document["job_id"],
+                run_id=document["run_id"],
+                backend=document["backend"],
+                repository=document["repository"],
+                job_name=document["job_name"],
+                labels=tuple(labels),
+            )
+        except KeyError as error:
+            raise ValueError(f"missing runner request field: {error.args[0]}") from error
+        if (
+            not _positive_int(request.job_id)
+            or not _positive_int(request.run_id)
+            or request.backend not in BACKENDS
+            or not _nonempty(request.repository)
+            or not _nonempty(request.job_name)
+            or REQUIRED_LABEL not in request.labels
+            or request.backend not in request.labels
+        ):
+            raise ValueError("invalid runner request")
+        return request
+
+
+@dataclass(frozen=True, slots=True)
+class IdentityBinding:
+    """The actual task claimed by an ephemeral runner."""
+
+    job_id: int
+    run_id: int
+    backend: str
+    runner_name: str
+    repository: str
+    job_name: str
+    schema_version: int = 1
+
+    @classmethod
+    def from_webhook(cls, payload: object) -> IdentityBinding | None:
+        if not isinstance(payload, dict) or payload.get("action") != "in_progress":
+            return None
+        job = payload.get("workflow_job")
+        repository = payload.get("repository")
+        if not isinstance(job, dict) or not isinstance(repository, dict):
+            return None
+
+        labels = job.get("labels")
+        if not isinstance(labels, list) or not all(
+            isinstance(label, str) for label in labels
+        ):
+            return None
+        selected = BACKENDS.intersection(labels)
+        if REQUIRED_LABEL not in labels or len(selected) != 1:
+            return None
+        backend = next(iter(selected))
+
+        job_id = job.get("id")
+        run_id = job.get("run_id")
+        runner_name = job.get("runner_name")
+        job_name = job.get("name")
+        full_name = repository.get("full_name")
+        if not _positive_int(job_id) or not _positive_int(run_id):
+            return None
+        if not all(
+            _nonempty(value)
+            for value in (runner_name, job_name, full_name)
+        ):
+            return None
+        if not runner_name.startswith(f"gitea-{backend}-"):
+            return None
+
+        return cls(
+            job_id=job_id,
+            run_id=run_id,
+            backend=backend,
+            runner_name=runner_name.strip(),
+            repository=full_name.strip(),
+            job_name=job_name.strip(),
+        )
+
+    def to_json(self) -> bytes:
+        return json.dumps(
+            asdict(self),
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode()
+
+
+def _positive_int(value: object) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool) and value > 0
+
+
+def _nonempty(value: object) -> bool:
+    return isinstance(value, str) and bool(value.strip())

src/gitea_dynamic_runner/pod_worker.py

@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+"""JetStream worker that creates one disposable Kubernetes Pod per job."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import re
+import ssl
+import uuid
+
+import nats
+from aiohttp import ClientResponseError
+from nats.errors import TimeoutError
+from nats.js.api import AckPolicy, ConsumerConfig
+
+from .kubernetes import KubernetesClient
+from .models import IdentityBinding, RunnerRequest
+
+
+LOG = logging.getLogger(__name__)
+STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+REQUEST_SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.pod")
+BINDING_SUBJECT = os.environ.get("NATS_BINDING_SUBJECT", "ci.runner.pod.binding")
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_USER = os.environ.get("NATS_USER", "ci-worker")
+NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/run/secrets/nats/password"))
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+KUBERNETES_API = os.environ.get("KUBERNETES_API", "https://kubernetes.default.svc")
+KUBERNETES_TOKEN_FILE = Path(
+    os.environ.get(
+        "KUBERNETES_TOKEN_FILE",
+        "/var/run/secrets/kubernetes.io/serviceaccount/token",
+    )
+)
+KUBERNETES_CA_FILE = Path(
+    os.environ.get(
+        "KUBERNETES_CA_FILE",
+        "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+    )
+)
+NAMESPACE = os.environ.get("RUNNER_NAMESPACE", "gitea-actions")
+RUNNER_IMAGE = os.environ.get("RUNNER_IMAGE", "docker.io/gitea/runner:2")
+RUNNER_SERVICE_ACCOUNT = os.environ.get("RUNNER_SERVICE_ACCOUNT", "gitea-dynamic-runner")
+RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runner")
+GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
+CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
+POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60)))
+SPIFFE_PATH_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$")
+
+
+def identity_path(repository: str, job_name: str) -> str:
+    parts = repository.split("/")
+    if len(parts) != 2 or not all(parts):
+        raise ValueError("repository must be owner/name")
+    if not job_name.strip():
+        raise ValueError("identity components must not be empty")
+    return "/".join(safe_identity_segment(part) for part in (*parts, job_name))
+
+
+def safe_identity_segment(value: str) -> str:
+    """Map arbitrary Gitea names to stable SPIFFE Operator path segments."""
+    if SPIFFE_PATH_SEGMENT.fullmatch(value):
+        return value
+    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-._")
+    slug = slug[:48].rstrip("-._") or "segment"
+    digest = hashlib.sha256(value.encode()).hexdigest()[:12]
+    return f"{slug}-{digest}"
+
+
+def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]:
+    if request.backend != "pod":
+        raise ValueError("Pod backend only accepts pod requests")
+    return {
+        "apiVersion": "v1",
+        "kind": "Pod",
+        "metadata": {
+            "name": pod_name,
+            "namespace": NAMESPACE,
+            "labels": {
+                "app.kubernetes.io/name": "gitea-dynamic-runner",
+                "app.kubernetes.io/component": "runner",
+                "ci.ddupan.top/backend": "pod",
+            },
+            "annotations": {
+                "ci.ddupan.top/queued-job-id": str(request.job_id),
+                "ci.ddupan.top/queued-run-id": str(request.run_id),
+            },
+        },
+        "spec": {
+            "serviceAccountName": RUNNER_SERVICE_ACCOUNT,
+            "restartPolicy": "Never",
+            "terminationGracePeriodSeconds": 30,
+            "containers": [
+                {
+                    "name": "runner",
+                    "image": RUNNER_IMAGE,
+                    "imagePullPolicy": "IfNotPresent",
+                    "securityContext": {"privileged": True},
+                    "env": [
+                        {"name": "GITEA_INSTANCE_URL", "value": GITEA_INSTANCE},
+                        {
+                            "name": "GITEA_RUNNER_NAME",
+                            "valueFrom": {"fieldRef": {"fieldPath": "metadata.name"}},
+                        },
+                        {
+                            "name": "GITEA_RUNNER_REGISTRATION_TOKEN_FILE",
+                            "value": "/run/secrets/gitea/token",
+                        },
+                        {"name": "GITEA_RUNNER_LABELS", "value": "self-hosted:host,pod:host"},
+                        {"name": "GITEA_RUNNER_EPHEMERAL", "value": "1"},
+                        {"name": "GITEA_RUNNER_ONCE", "value": "1"},
+                        {"name": "CONFIG_FILE", "value": "/etc/gitea-runner/config.yaml"},
+                    ],
+                    "volumeMounts": [
+                        {
+                            "name": "registration-token",
+                            "mountPath": "/run/secrets/gitea",
+                            "readOnly": True,
+                        },
+                        {
+                            "name": "spire-agent-socket",
+                            "mountPath": "/run/spire/agent-sockets",
+                            "readOnly": True,
+                        },
+                    ],
+                }
+            ],
+            "volumes": [
+                {
+                    "name": "registration-token",
+                    "secret": {
+                        "secretName": RUNNER_TOKEN_SECRET,
+                        "items": [{"key": "token", "path": "token"}],
+                    },
+                },
+                {
+                    "name": "spire-agent-socket",
+                    "csi": {"driver": "csi.spiffe.io", "readOnly": True},
+                },
+            ],
+        },
+    }
+
+
+async def heartbeat(message: object, stop: asyncio.Event) -> None:
+    while True:
+        try:
+            await asyncio.wait_for(stop.wait(), timeout=60)
+            return
+        except asyncio.TimeoutError:
+            await message.in_progress()
+
+
+async def wait_for_pod(client: KubernetesClient, name: str) -> bool:
+    deadline = asyncio.get_running_loop().time() + POD_TIMEOUT
+    while asyncio.get_running_loop().time() < deadline:
+        pod = await client.get_pod(name)
+        if pod is None:
+            raise RuntimeError(f"runner Pod {name} disappeared")
+        status = pod.get("status")
+        phase = status.get("phase") if isinstance(status, dict) else None
+        if phase == "Succeeded":
+            return True
+        if phase == "Failed":
+            return False
+        await asyncio.sleep(2)
+    raise asyncio.TimeoutError(f"runner Pod {name} timed out")
+
+
+async def run_request(message: object, client: KubernetesClient) -> None:
+    try:
+        request = RunnerRequest.from_json(message.data)
+    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
+        LOG.error("discarding invalid Pod request: %s", error)
+        await message.ack()
+        return
+    if request.backend != "pod":
+        LOG.error("discarding %s request received by Pod worker", request.backend)
+        await message.ack()
+        return
+
+    pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}"
+    LOG.info(
+        "runner request received job_id=%d run_id=%d repository=%s job_name=%r "
+        "backend=%s pod=%s",
+        request.job_id,
+        request.run_id,
+        request.repository,
+        request.job_name,
+        request.backend,
+        pod_name,
+    )
+    stop = asyncio.Event()
+    pulse = asyncio.create_task(heartbeat(message, stop))
+    try:
+        await client.create_pod(pod_manifest(request, pod_name))
+        LOG.info(
+            "runner Pod created pod=%s job_id=%d run_id=%d image=%s",
+            pod_name,
+            request.job_id,
+            request.run_id,
+            RUNNER_IMAGE,
+        )
+        if await wait_for_pod(client, pod_name):
+            await message.ack()
+            LOG.info(
+                "runner request acknowledged pod=%s job_id=%d result=succeeded",
+                pod_name,
+                request.job_id,
+            )
+        else:
+            LOG.error(
+                "runner Pod failed pod=%s job_id=%d; request will be retried",
+                pod_name,
+                request.job_id,
+            )
+            await message.nak(delay=30)
+    except Exception:
+        LOG.exception(
+            "runner request failed pod=%s job_id=%d; request will be retried",
+            pod_name,
+            request.job_id,
+        )
+        await message.nak(delay=30)
+        raise
+    finally:
+        stop.set()
+        await pulse
+        LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
+        await client.delete_pod(pod_name)
+
+
+async def bind_request(message: object, client: KubernetesClient) -> None:
+    binding: IdentityBinding | None = None
+    try:
+        document = json.loads(message.data)
+        binding = IdentityBinding(**document)
+        if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
+            raise ValueError("invalid Pod identity binding")
+        path = identity_path(binding.repository, binding.job_name)
+        LOG.info(
+            "identity binding received job_id=%d run_id=%d runner_name=%s "
+            "repository=%s job_name=%r identity_path=%s",
+            binding.job_id,
+            binding.run_id,
+            binding.runner_name,
+            binding.repository,
+            binding.job_name,
+            path,
+        )
+        await client.bind_identity(binding.runner_name, path)
+    except ClientResponseError as error:
+        if error.status == 404:
+            LOG.warning(
+                "identity binding Pod not found runner_name=%s job_id=%s; binding will be retried",
+                binding.runner_name if binding else None,
+                binding.job_id if binding else None,
+            )
+            await message.nak(delay=2)
+            return
+        LOG.exception(
+            "identity binding Kubernetes request failed runner_name=%s job_id=%s status=%d",
+            binding.runner_name if binding else None,
+            binding.job_id if binding else None,
+            error.status,
+        )
+        await message.nak(delay=30)
+        raise
+    except (json.JSONDecodeError, TypeError, ValueError) as error:
+        LOG.error("discarding invalid identity binding: %s", error)
+        await message.ack()
+        return
+    await message.ack()
+    LOG.info(
+        "identity binding applied and acknowledged runner_name=%s job_id=%d identity_path=%s",
+        binding.runner_name,
+        binding.job_id,
+        path,
+    )
+
+
+async def consume_requests(subscription: object, client: KubernetesClient) -> None:
+    active: set[asyncio.Task[None]] = set()
+    while True:
+        active = {task for task in active if not task.done()}
+        free = CAPACITY - len(active)
+        if free < 1:
+            await asyncio.wait(active, return_when=asyncio.FIRST_COMPLETED)
+            continue
+        try:
+            messages = await subscription.fetch(batch=free, timeout=5)
+        except TimeoutError:
+            continue
+        for message in messages:
+            task = asyncio.create_task(run_request(message, client))
+            task.add_done_callback(_report_task)
+            active.add(task)
+
+
+async def consume_bindings(subscription: object, client: KubernetesClient) -> None:
+    while True:
+        try:
+            messages = await subscription.fetch(batch=16, timeout=5)
+        except TimeoutError:
+            continue
+        await asyncio.gather(*(bind_request(message, client) for message in messages))
+
+
+def _report_task(task: asyncio.Task[None]) -> None:
+    if not task.cancelled() and (error := task.exception()) is not None:
+        LOG.error("Pod runner task failed", exc_info=(type(error), error, error.__traceback__))
+
+
+async def main() -> None:
+    if CAPACITY < 1:
+        raise ValueError("RUNNER_CAPACITY must be at least 1")
+    context = ssl.create_default_context(cafile=NATS_CA_FILE)
+    nc = await nats.connect(
+        NATS_URL,
+        user=NATS_USER,
+        password=NATS_PASSWORD_FILE.read_text().strip(),
+        tls=context,
+        name="pod-runner-worker",
+    )
+    js = nc.jetstream()
+    requests = await js.pull_subscribe(
+        REQUEST_SUBJECT,
+        durable="pod",
+        stream=STREAM,
+        config=ConsumerConfig(
+            durable_name="pod",
+            filter_subject=REQUEST_SUBJECT,
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=5 * 60,
+            max_ack_pending=max(CAPACITY, 1),
+            max_deliver=5,
+        ),
+    )
+    bindings = await js.pull_subscribe(
+        BINDING_SUBJECT,
+        durable="pod-binding",
+        stream=STREAM,
+        config=ConsumerConfig(
+            durable_name="pod-binding",
+            filter_subject=BINDING_SUBJECT,
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=30,
+            max_ack_pending=64,
+            max_deliver=10,
+        ),
+    )
+    async with KubernetesClient(
+        api_url=KUBERNETES_API,
+        token_file=KUBERNETES_TOKEN_FILE,
+        ca_file=KUBERNETES_CA_FILE,
+        namespace=NAMESPACE,
+    ) as client:
+        try:
+            await asyncio.gather(
+                consume_requests(requests, client),
+                consume_bindings(bindings, client),
+            )
+        finally:
+            await nc.drain()
+
+
+def cli() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    cli()

src/gitea_dynamic_runner/worker.py

@@ -2,6 +2,7 @@
 """Capacity-bounded JetStream consumer that launches one ephemeral VM per job."""
 
 import asyncio
+import json
 import logging
 import os
 import secrets
@@ -14,11 +15,13 @@ from aiohttp import web
 from nats.errors import TimeoutError
 from nats.js.api import AckPolicy, ConsumerConfig
 
+from .models import RunnerRequest
+
 LOG = logging.getLogger(__name__)
 CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "1"))
-SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.kind-microvm")
+SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm")
 STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
-DURABLE = os.environ.get("NATS_DURABLE", "kind-microvm")
+DURABLE = os.environ.get("NATS_DURABLE", "vm")
 MAX_INFLIGHT = int(os.environ.get("RUNNER_MAX_INFLIGHT", "64"))
 NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
 NATS_USER = os.environ.get("NATS_USER", "ci-worker")
@@ -28,6 +31,9 @@ REGISTRATION_TOKEN_FILE = Path(os.environ.get("REGISTRATION_TOKEN_FILE", "/etc/m
 LAUNCHER = os.environ.get("LAUNCHER", "/usr/local/libexec/microvm-runner-launch")
 TOKEN_LISTEN = os.environ.get("TOKEN_LISTEN", "172.30.0.1")
 TOKEN_PORT = int(os.environ.get("TOKEN_PORT", "8787"))
+GUEST_ASSETS = Path(
+    os.environ.get("GUEST_ASSETS", "/opt/gitea-dynamic-runner/guest-assets.tar.gz")
+)
 
 tokens: dict[str, bytes] = {}
 token_lock = asyncio.Lock()
@@ -42,6 +48,13 @@ async def token(request: web.Request) -> web.Response:
     return web.Response(body=value, headers={"Cache-Control": "no-store"})
 
 
+async def guest_assets(_: web.Request) -> web.FileResponse:
+    return web.FileResponse(
+        GUEST_ASSETS,
+        headers={"Cache-Control": "public, immutable"},
+    )
+
+
 async def heartbeat(message: object, stop: asyncio.Event) -> None:
     while True:
         try:
@@ -52,6 +65,17 @@ async def heartbeat(message: object, stop: asyncio.Event) -> None:
 
 
 async def run_one(message: object) -> None:
+    try:
+        request = RunnerRequest.from_json(message.data)
+    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
+        LOG.error("discarding invalid runner request: %s", error)
+        await message.ack()
+        return
+    if request.backend != "vm":
+        LOG.error("discarding %s request received by VM worker", request.backend)
+        await message.ack()
+        return
+
     instance_id = str(uuid.uuid4())
     nonce = secrets.token_urlsafe(32)
     async with token_lock:
@@ -59,7 +83,18 @@ async def run_one(message: object) -> None:
     stop = asyncio.Event()
     pulse = asyncio.create_task(heartbeat(message, stop))
     try:
-        process = await asyncio.create_subprocess_exec(LAUNCHER, instance_id, nonce)
+        process = await asyncio.create_subprocess_exec(
+            LAUNCHER,
+            instance_id,
+            nonce,
+            env={
+                **os.environ,
+                "RUNNER_JOB_ID": str(request.job_id),
+                "RUNNER_RUN_ID": str(request.run_id),
+                "RUNNER_REPOSITORY": request.repository,
+                "RUNNER_JOB_NAME": request.job_name,
+            },
+        )
         return_code = await process.wait()
         if return_code == 0:
             await message.ack()
@@ -137,6 +172,7 @@ async def consume() -> None:
 async def main() -> None:
     app = web.Application()
     app.router.add_get("/token/{nonce}", token)
+    app.router.add_get("/assets/guest-assets.tar.gz", guest_assets)
     runner = web.AppRunner(app)
     await runner.setup()
     await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start()

src/gitea_microvm_runner/__init__.py

@@ -1 +0,0 @@
-"""Gitea microVM runner controller and worker."""

systemd/gitea-microvm-worker.service

@@ -7,7 +7,7 @@ Requires=microvm-runner-network.service
 [Service]
 Type=simple
 EnvironmentFile=-/etc/microvm-runner/worker.env
-ExecStart=/opt/gitea-microvm-runner/venv/bin/gitea-microvm-worker
+ExecStart=/opt/gitea-dynamic-runner/venv/bin/gitea-dynamic-runner-vm-worker
 Restart=on-failure
 RestartSec=5s
 SupplementaryGroups=kvm

tests/test_controller.py

@@ -1,22 +1,88 @@
 import hashlib
 import hmac
+import json
 
-from gitea_microvm_runner import controller
+import pytest
+
+from gitea_dynamic_runner import controller
+from gitea_dynamic_runner.models import IdentityBinding, RunnerRequest
 
 
-def test_accepts_matching_queued_job(monkeypatch):
+def queued_job(**overrides):
-    monkeypatch.setattr(controller, "LABEL", "kind-microvm")
+    job = {
-    assert controller.accepts({
+        "id": 47,
+        "run_id": 12,
+        "name": "publish-image",
+        "labels": ["self-hosted", "pod"],
+    }
+    job.update(overrides)
+    return {
         "action": "queued",
-        "workflow_job": {"id": 47, "labels": ["linux", "kind-microvm"]},
+        "workflow_job": job,
-    }) == (True, "47")
+        "repository": {"full_name": "panxiao81/example"},
+    }
 
 
-def test_rejects_other_actions_labels_and_boolean_id(monkeypatch):
+def test_accepts_matching_queued_job():
-    monkeypatch.setattr(controller, "LABEL", "kind-microvm")
+    assert controller.accepts(queued_job()) == (True, "47")
-    assert controller.accepts({"action": "completed", "workflow_job": {"id": 1, "labels": ["kind-microvm"]}}) == (False, None)
+
-    assert controller.accepts({"action": "queued", "workflow_job": {"id": 1, "labels": ["host"]}}) == (False, None)
+
-    assert controller.accepts({"action": "queued", "workflow_job": {"id": True, "labels": ["kind-microvm"]}}) == (False, None)
+def test_rejects_other_actions_labels_and_boolean_id():
+    completed = queued_job()
+    completed["action"] = "completed"
+    assert controller.accepts(completed) == (False, None)
+    assert controller.accepts(queued_job(labels=["self-hosted", "other"])) == (False, None)
+    assert controller.accepts(queued_job(labels=["self-hosted", "pod", "vm"])) == (False, None)
+    assert controller.accepts(queued_job(id=True)) == (False, None)
+
+
+def test_runner_request_contains_stable_identity_context():
+    request = RunnerRequest.from_webhook(queued_job())
+    assert request is not None
+    assert request.backend == "pod"
+    assert request.repository == "panxiao81/example"
+    assert request.job_name == "publish-image"
+    assert request.job_id == 47
+    assert json.loads(request.to_json()) == {
+        "backend": "pod",
+        "job_id": 47,
+        "job_name": "publish-image",
+        "labels": ["self-hosted", "pod"],
+        "repository": "panxiao81/example",
+        "run_id": 12,
+        "schema_version": 1,
+    }
+
+
+def test_runner_request_requires_complete_identity_context():
+    assert RunnerRequest.from_webhook(queued_job(run_id=None)) is None
+    assert RunnerRequest.from_webhook(queued_job(name=" ")) is None
+
+
+def test_runner_request_json_round_trip_and_validation():
+    request = RunnerRequest.from_webhook(queued_job())
+    assert request is not None
+    assert RunnerRequest.from_json(request.to_json()) == request
+    with pytest.raises(ValueError, match="unsupported"):
+        RunnerRequest.from_json(b'{"schema_version":2}')
+
+
+def test_identity_binding_uses_actual_runner_assignment():
+    payload = queued_job(runner_name="gitea-pod-6c47d03d")
+    payload["action"] = "in_progress"
+    binding = IdentityBinding.from_webhook(payload)
+    assert binding is not None
+    assert binding.backend == "pod"
+    assert binding.runner_name == "gitea-pod-6c47d03d"
+    assert binding.repository == "panxiao81/example"
+    assert binding.job_name == "publish-image"
+    assert json.loads(binding.to_json())["job_id"] == 47
+
+
+def test_identity_binding_rejects_runner_from_another_pool():
+    payload = queued_job(runner_name="gitea-vm-6c47d03d")
+    payload["action"] = "in_progress"
+    assert IdentityBinding.from_webhook(payload) is None
 
 
 def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):

tests/test_jwt_broker.py

@@ -0,0 +1,16 @@
+import pytest
+
+from gitea_dynamic_runner.jwt_broker import extract_svid
+
+
+def test_extract_svid_from_spire_json():
+    assert extract_svid([
+        {"svids": [{"hint": "", "spiffe_id": "spiffe://example/ci", "svid": "jwt"}]},
+        {"bundles": {}},
+    ]) == "jwt"
+
+
+@pytest.mark.parametrize("document", [{}, [], [{"bundles": {}}]])
+def test_extract_svid_rejects_unexpected_response(document):
+    with pytest.raises(ValueError):
+        extract_svid(document)

tests/test_pod_worker.py

@@ -0,0 +1,57 @@
+import pytest
+
+from gitea_dynamic_runner.models import RunnerRequest
+from gitea_dynamic_runner import pod_worker
+
+
+def request() -> RunnerRequest:
+    return RunnerRequest(
+        job_id=47,
+        run_id=12,
+        backend="pod",
+        repository="panxiao81/example",
+        job_name="publish image",
+        labels=("self-hosted", "pod"),
+    )
+
+
+def test_identity_path_is_meaningful_and_uri_safe():
+    path = pod_worker.identity_path("panxiao81/example", "publish image")
+    assert path.startswith("panxiao81/example/publish-image-")
+    assert "%" not in path
+    assert all(pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part) for part in path.split("/"))
+    assert pod_worker.identity_path("panxiao81/example", "lint") == "panxiao81/example/lint"
+    assert path == pod_worker.identity_path("panxiao81/example", "publish image")
+    assert path != pod_worker.identity_path("panxiao81/example", "publish-image")
+    real_path = pod_worker.identity_path(
+        "panxiao81/postgresql-tenant-operator", "Run on Ubuntu"
+    )
+    assert real_path.startswith(
+        "panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-"
+    )
+    assert all(
+        pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part)
+        for part in real_path.split("/")
+    )
+    with pytest.raises(ValueError):
+        pod_worker.identity_path("invalid", "test")
+
+
+def test_pod_manifest_is_disposable_host_runner(monkeypatch):
+    monkeypatch.setattr(pod_worker, "NAMESPACE", "ci")
+    manifest = pod_worker.pod_manifest(request(), "gitea-pod-abcd")
+    assert manifest["metadata"]["name"] == "gitea-pod-abcd"
+    assert manifest["metadata"]["annotations"]["ci.ddupan.top/queued-job-id"] == "47"
+    spec = manifest["spec"]
+    assert spec["restartPolicy"] == "Never"
+    container = spec["containers"][0]
+    assert container["securityContext"] == {"privileged": True}
+    env = {item["name"]: item for item in container["env"]}
+    assert env["GITEA_RUNNER_LABELS"]["value"] == "self-hosted:host,pod:host"
+    assert env["GITEA_RUNNER_EPHEMERAL"]["value"] == "1"
+    assert env["GITEA_RUNNER_ONCE"]["value"] == "1"
+    assert all(item["name"] != "DOCKER_HOST" for item in container["env"])
+    volumes = {item["name"]: item for item in spec["volumes"]}
+    assert volumes["spire-agent-socket"]["csi"]["driver"] == "csi.spiffe.io"
+    assert "runner-config" not in volumes
+    assert "docker" not in volumes