确认完成事件后 ACK 持久消息

Reviewed-on: #17

.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,122 @@
+---
+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, vm]
+    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: 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)
+          cleanup() {
+            docker buildx rm ci-builder >/dev/null 2>&1 || true
+            rm -rf -- "$docker_config" "$jwt_file"
+          }
+          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
+
+          docker buildx create \
+            --name ci-builder \
+            --driver docker-container \
+            --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/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

@@ -17,25 +17,23 @@ runs-on: [self-hosted, vm]
 
 组件：
 
-- `controller`：接收 Gitea `workflow_job` webhook，将指定 label 的 queued job
-  发布到 NATS JetStream。
-- `worker`：领取任务、限制并发，并通过 Pod 或 microVM backend 创建一次性环境。
-- `microvm-runner-launch`：为每个任务创建 COW disk、NoCloud seed 和 TAP，运行
+- `controller`：接收 Gitea `workflow_job` webhook，直接调用 OpenSandbox Lifecycle
+  API，从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。
+- `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行
   Cloud Hypervisor，退出后完整清理。
 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral
   runner，执行一个 job 后关机。
+- `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 使用
   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；路线与迁移边界见
+OpenSandbox 路径不使用 NATS。旧 Pod/microVM worker 只作为迁移期代码保留，不应重新
+部署。长期 RunnerService 协议路线见
 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
 
 ## 开发
@@ -49,11 +47,18 @@ pytest
 
 ## 安全边界
 
-- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。
+- OpenSandbox API key、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: []

container/runner.Dockerfile

@@ -1,8 +1,21 @@
 FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
 
-FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242
+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 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
-RUN chmod 0755 /usr/local/libexec/gitea-job-started
+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"]

docs/opensandbox-runner.md

@@ -0,0 +1,52 @@
+# OpenSandbox runner
+
+Gitea 的 `workflow_job` webhook 只负责发现带 `self-hosted,pod` 或
+`self-hosted,vm` label 的 queued job。controller 不再把请求写入 NATS，而是携带由
+homelab ExternalSecret 挂载的 API key 直接调用 OpenSandbox Lifecycle API：
+
+```text
+http://10.60.0.13:8080/v1/sandboxes
+```
+
+请求通过 `extensions.poolRef` 选择 `ci-pod` 或 `ci-vm`。地址是 VyOS HAProxy 的
+内网 TCP frontend，backend 为 sandbox 两个节点上的固定 NodePort；不通过公网或
+Cloudflare Tunnel。OpenSandbox key 只从 OpenBao `kv/k8s/opensandbox-api.api_key`
+进入 homelab Secret，进程通过 `OPENSANDBOX_API_KEY_FILE` 读取。
+
+## 身份顺序
+
+Lifecycle 请求把稳定的 repository/task SPIFFE ID 放入 task environment。sandbox
+集群内的 `opensandbox-identity` controller 读取 BatchSandbox allocation 得到实际
+Pod UID，然后创建：
+
+- parent：`spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/pod/<pod-uid>`；
+- workload：`spiffe://ddupan.top/ci/<owner>/<repository>/<task>`；
+- selector：`unix:uid:2000`。
+
+job ID 只进入诊断 label，不进入业务身份。Pool 内 runner 进程固定使用 UID 2000，
+Pod 设置 `shareProcessNamespace: true`；guest-local SPIRE Agent 使用 Pod-bound PSAT，
+通过内存 emptyDir 暴露 Workload API。`gitea-opensandbox-runner` 等待精确 SVID，随后
+通过一次性 nonce URL 领取 Gitea registration token。token 不进入 Lifecycle 请求、
+BatchSandbox、镜像或 sandbox Secret。
+
+identity controller 只能读取 opensandbox namespace 的 BatchSandbox/Pod，并维护带
+自身 label 的 ClusterStaticEntry。它不持有 OpenSandbox API key、Gitea token 或 Bao
+凭据。BatchSandbox 消失后，对应 entry 在下一次 reconcile 删除；Pod 删除后，按 Pod
+UID attestation 的临时 Agent 失去父级。
+
+## Pool 契约
+
+`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；
+- `self-hosted` 必须是所有 runner labels 的前缀；
+- ephemeral/once runner 完成一项任务后退出。
+
+## 清理与恢复
+
+controller 监控 Lifecycle 状态，在终止、失败、超时或取消时调用 DELETE。API delete、
+identity entry delete 均接受对象已不存在。controller 重启时，OpenSandbox timeout
+仍是最终回收边界；后续可基于 metadata list 恢复主动监控，但不得为此重新引入消息
+队列。

pyproject.toml

@@ -16,6 +16,7 @@ test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
 gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main"
 gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli"
 gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli"
+gitea-dynamic-runner-opensandbox-identity = "gitea_dynamic_runner.opensandbox_identity:cli"
 gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
 
 [tool.pytest.ini_options]

scripts/gitea-microvm-guest-runner

@@ -6,6 +6,8 @@ instance=${2:?Gitea instance is required}
 runner_name=${3:?runner name 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"
@@ -25,4 +27,5 @@ gitea-runner register \
     --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/gitea-opensandbox-runner

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+: "${CI_SPIFFE_ID:?CI_SPIFFE_ID is required}"
+: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
+: "${GITEA_RUNNER_REGISTRATION_TOKEN_URL:?GITEA_RUNNER_REGISTRATION_TOKEN_URL is required}"
+
+socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
+token_file=$(mktemp)
+trap 'rm -f -- "$token_file"' EXIT
+export GITEA_RUNNER_REGISTRATION_TOKEN_FILE="$token_file"
+deadline=$((SECONDS + 120))
+while (( SECONDS < deadline )); do
+  if /opt/spire/bin/spire-agent api fetch x509 \
+    -socketPath "$socket_path" \
+    -output json 2>/dev/null | \
+    jq -e --arg id "$CI_SPIFFE_ID" \
+      'any(.svids[]?; .spiffe_id == $id)' >/dev/null; then
+    umask 077
+    curl --fail --silent --show-error --retry 10 --retry-all-errors \
+      --connect-timeout 2 --max-time 30 \
+      "$GITEA_RUNNER_REGISTRATION_TOKEN_URL" >"$token_file"
+    /usr/local/bin/run.sh
+    exit $?
+  fi
+  sleep 1
+done
+
+printf 'timed out waiting for OpenSandbox SPIFFE identity %s\n' "$CI_SPIFFE_ID" >&2
+exit 1

scripts/microvm-runner-launch

@@ -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_labels" ]
+  - [ 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" \
-    --firmware "$firmware" \
-    --cpus "boot=$cpus" \
-    --memory "size=$memory" \
-    --disk "path=$overlay" \
-    --disk "path=$seed,readonly=on" \
-    --net "tap=$tap,mac=$mac" \
-    --serial "file=$serial" \
-    --console off
+if timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
+        --firmware "$firmware" \
+        --cpus "boot=$cpus" \
+        --memory "size=$memory,shared=on" \
+        --disk "path=$overlay,image_type=qcow2,direct=on,sparse=off" \
+        --disk "path=$seed,readonly=on,image_type=raw,direct=on,sparse=off" \
+        --net "tap=$tap,mac=$mac" \
+        --serial "file=$serial" \
+        --console off; then
+    grep -Fq gitea-runner-job-complete "$serial"
+else
+    exit $?
+fi

src/gitea_dynamic_runner/controller.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python3
-"""Gitea workflow_job webhook to NATS JetStream producer."""
+"""Gitea workflow_job webhook to the OpenSandbox Lifecycle API."""
 
+from __future__ import annotations
+
+import asyncio
 import hashlib
 import hmac
 import json
@@ -11,53 +14,63 @@ from pathlib import Path
 
 import nats
 from aiohttp import web
-from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
+from nats.errors import TimeoutError as NatsTimeoutError
+from nats.js.api import (
+    AckPolicy,
+    ConsumerConfig,
+    DiscardPolicy,
+    RetentionPolicy,
+    StorageType,
+    StreamConfig,
+)
 from nats.js.errors import NotFoundError
 
-from .models import IdentityBinding, RunnerRequest
+from .models import RunnerRequest
+from .opensandbox import OpenSandboxClient
+from .opensandbox_worker import OpenSandboxScheduler, RegistrationTokens
 
 
 LOG = logging.getLogger(__name__)
-SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
-STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+WEBHOOK_SECRET_FILE = Path(
+    os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")
+)
+REGISTRATION_TOKEN_FILE = Path(
+    os.environ.get("REGISTRATION_TOKEN_FILE", "/run/secrets/gitea/registration-token")
+)
+OPENSANDBOX_API = os.environ.get("OPENSANDBOX_API", "http://10.60.0.13:8080")
+OPENSANDBOX_API_KEY_FILE = Path(
+    os.environ.get(
+        "OPENSANDBOX_API_KEY_FILE",
+        "/run/secrets/opensandbox/api-key",
+    )
+)
 NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
-NATS_USER = os.environ.get("NATS_USER", "ci-producer")
-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")
-WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret"))
+NATS_PRODUCER_USER = os.environ.get("NATS_PRODUCER_USER", "ci-producer")
+NATS_PRODUCER_PASSWORD_FILE = Path(
+    os.environ.get("NATS_PRODUCER_PASSWORD_FILE", "/run/secrets/nats/producer-password")
+)
+NATS_WORKER_USER = os.environ.get("NATS_WORKER_USER", "ci-worker")
+NATS_WORKER_PASSWORD_FILE = Path(
+    os.environ.get("NATS_WORKER_PASSWORD_FILE", "/run/secrets/nats/worker-password")
+)
+NATS_STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+NATS_SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
+POD_CONSUMER_ENABLED = os.environ.get("POD_CONSUMER_ENABLED", "true") == "true"
 
 
 def accepts(payload: object) -> tuple[bool, str | None]:
-    """Compatibility helper for callers that only need acceptance and identity."""
     request = RunnerRequest.from_webhook(payload)
     return (request is not None, str(request.job_id) if request else None)
 
 
 def valid_signature(body: bytes, signature: str) -> bool:
-    expected = hmac.new(WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256).hexdigest()
+    expected = hmac.new(
+        WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256
+    ).hexdigest()
     return hmac.compare_digest(signature.removeprefix("sha256="), expected)
 
 
-async def ensure_stream(js: object) -> None:
-    config = StreamConfig(
-        name=STREAM,
-        subjects=[f"{SUBJECT_PREFIX}.>"],
-        retention=RetentionPolicy.WORK_QUEUE,
-        storage=StorageType.FILE,
-        discard=DiscardPolicy.OLD,
-        max_age=24 * 60 * 60,
-        max_msgs=10_000,
-        max_bytes=256 * 1024 * 1024,
-        duplicate_window=24 * 60 * 60,
-    )
-    try:
-        await js.stream_info(STREAM)
-    except NotFoundError:
-        await js.add_stream(config=config)
-    else:
-        await js.update_stream(config=config)
-
-
 async def webhook(request: web.Request) -> web.Response:
     body = await request.read()
     if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")):
@@ -69,50 +82,50 @@ async def webhook(request: web.Request) -> web.Response:
     context = _webhook_context(payload)
     LOG.info("workflow_job webhook received %s", context)
 
+    scheduler: OpenSandboxScheduler = request.app["scheduler"]
+    if isinstance(payload, dict) and payload.get("action") == "completed":
+        job = payload.get("workflow_job")
+        runner_name = job.get("runner_name") if isinstance(job, dict) else None
+        if isinstance(runner_name, str) and runner_name.startswith(
+            ("gitea-pod-", "gitea-vm-")
+        ):
+            cleaned = await scheduler.complete(runner_name)
+            await request.app["js"].publish(
+                f"{NATS_SUBJECT_PREFIX}.lifecycle.completed",
+                body,
+                headers={
+                    "Nats-Msg-Id": (f"gitea-workflow-job-{job.get('id')}-completed")
+                },
+            )
+            LOG.info(
+                "completed runner cleanup runner=%r cleaned=%s %s",
+                runner_name,
+                cleaned,
+                context,
+            )
+        return web.Response(status=204)
+
     runner_request = RunnerRequest.from_webhook(payload)
-    if runner_request is not None:
-        subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
-        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")
+    if runner_request is None:
+        LOG.info("workflow_job webhook ignored %s", context)
+        return web.Response(status=204)
 
-    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")
+    subject = f"{NATS_SUBJECT_PREFIX}.{runner_request.backend}"
+    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 persisted subject=%s %s", subject, context)
+    return web.Response(status=202, text="queued\n")
 
-    LOG.warning("workflow_job webhook ignored %s", context)
-    return web.Response(status=204)
+
+async def registration_token(request: web.Request) -> web.Response:
+    scheduler: OpenSandboxScheduler = request.app["scheduler"]
+    value = await scheduler.tokens.consume(request.match_info["nonce"])
+    if value is None:
+        raise web.HTTPNotFound()
+    return web.Response(body=value, headers={"Cache-Control": "no-store"})
 
 
 def _webhook_context(payload: object) -> str:
@@ -131,37 +144,152 @@ def _webhook_context(payload: object) -> str:
 
 
 async def health(request: web.Request) -> web.Response:
-    connected = request.app["nc"].is_connected
-    return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503)
-
-
-async def nats_context(app: web.Application):
-    tls = 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=tls,
-        name="microvm-runner-controller",
+    client: OpenSandboxClient = request.app["opensandbox_client"]
+    return web.Response(
+        text="ok\n" if client.session is not None else "disconnected\n",
+        status=200 if client.session is not None else 503,
     )
-    app["nc"] = nc
-    app["js"] = nc.jetstream()
+
+
+async def ensure_stream(js: object) -> None:
+    config = StreamConfig(
+        name=NATS_STREAM,
+        subjects=[f"{NATS_SUBJECT_PREFIX}.>"],
+        retention=RetentionPolicy.WORK_QUEUE,
+        storage=StorageType.FILE,
+        discard=DiscardPolicy.OLD,
+        max_age=24 * 60 * 60,
+        max_msgs=10_000,
+        max_bytes=256 * 1024 * 1024,
+        duplicate_window=24 * 60 * 60,
+    )
+    try:
+        await js.stream_info(NATS_STREAM)
+    except NotFoundError:
+        await js.add_stream(config=config)
+    else:
+        await js.update_stream(config=config)
+
+
+async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
+    try:
+        runner_request = RunnerRequest.from_json(message.data)
+        await scheduler.create(runner_request)
+        task = scheduler.active[runner_request.job_id]
+        try:
+            while not task.done():
+                try:
+                    await asyncio.wait_for(asyncio.shield(task), timeout=30)
+                except asyncio.TimeoutError:
+                    await message.in_progress()
+            await task
+        except asyncio.CancelledError:
+            # A completed webhook cancels the lifecycle monitor after its
+            # Lifecycle DELETE succeeds. That is successful message handling.
+            # During controller shutdown the scheduler still owns the job, so
+            # preserve the unacked message for redelivery.
+            if runner_request.job_id in scheduler.active:
+                raise
+    except ValueError as error:
+        LOG.info("runner request already active: %s", error)
+        await message.nak(delay=5)
+    except Exception:
+        LOG.exception("persistent runner request failed")
+        await message.nak(delay=15)
+    else:
+        await message.ack()
+
+
+async def consume_pod_requests(js: object, scheduler: OpenSandboxScheduler) -> None:
+    subscription = await js.pull_subscribe(
+        f"{NATS_SUBJECT_PREFIX}.pod",
+        # Reuse the existing durable so no pending Pod request is stranded
+        # during the in-process consumer migration.
+        durable="pod",
+        stream=NATS_STREAM,
+        config=ConsumerConfig(
+            durable_name="pod",
+            filter_subject=f"{NATS_SUBJECT_PREFIX}.pod",
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=5 * 60,
+            max_ack_pending=4,
+            max_deliver=20,
+        ),
+    )
+    active: set[asyncio.Task[None]] = set()
+    while True:
+        active = {task for task in active if not task.done()}
+        try:
+            messages = await subscription.fetch(
+                batch=max(1, 4 - len(active)), timeout=5
+            )
+        except NatsTimeoutError:
+            continue
+        for message in messages:
+            task = asyncio.create_task(run_message(message, scheduler))
+            active.add(task)
+
+
+async def runtime_context(app: web.Application):
+    tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
+    tls = ssl.create_default_context(cafile=NATS_CA_FILE)
+    producer = await nats.connect(
+        NATS_URL,
+        user=NATS_PRODUCER_USER,
+        password=NATS_PRODUCER_PASSWORD_FILE.read_text().strip(),
+        tls=tls,
+        name="opensandbox-runner-controller-producer",
+    )
+    worker = await nats.connect(
+        NATS_URL,
+        user=NATS_WORKER_USER,
+        password=NATS_WORKER_PASSWORD_FILE.read_text().strip(),
+        tls=ssl.create_default_context(cafile=NATS_CA_FILE),
+        name="opensandbox-runner-controller-worker",
+    )
+    app["nc"] = producer
+    app["js"] = producer.jetstream()
     await ensure_stream(app["js"])
-    yield
-    await nc.drain()
+    async with OpenSandboxClient(
+        api_url=OPENSANDBOX_API,
+        api_key_file=OPENSANDBOX_API_KEY_FILE,
+    ) as client:
+        scheduler = OpenSandboxScheduler(client, tokens)
+        app["opensandbox_client"] = client
+        app["scheduler"] = scheduler
+        consumer = None
+        if POD_CONSUMER_ENABLED:
+            consumer = asyncio.create_task(
+                consume_pod_requests(worker.jetstream(), scheduler),
+                name="opensandbox-pod-consumer",
+            )
+        try:
+            yield
+        finally:
+            if consumer is not None:
+                consumer.cancel()
+                await asyncio.gather(consumer, return_exceptions=True)
+            await scheduler.close()
+            await worker.drain()
+            await producer.drain()
 
 
 def create_app() -> web.Application:
     app = web.Application(client_max_size=1024 * 1024)
-    app.cleanup_ctx.append(nats_context)
+    app.cleanup_ctx.append(runtime_context)
     app.router.add_post("/webhook", webhook)
+    app.router.add_get("/token/{nonce}", registration_token)
     app.router.add_get("/healthz", health)
     return app
 
 
 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")))
+    web.run_app(
+        create_app(),
+        host=os.environ.get("LISTEN", "0.0.0.0"),
+        port=int(os.environ.get("PORT", "8787")),
+    )
 
 
 if __name__ == "__main__":

src/gitea_dynamic_runner/opensandbox.py

@@ -0,0 +1,79 @@
+"""Minimal client for the OpenSandbox lifecycle API."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from urllib.parse import quote
+
+from aiohttp import ClientResponseError, ClientSession
+
+
+class OpenSandboxClient:
+    def __init__(
+        self,
+        *,
+        api_url: str,
+        api_key_file: Path | None = None,
+    ) -> None:
+        self.api_url = api_url.rstrip("/")
+        self.api_key_file = api_key_file
+        self.session: ClientSession | None = None
+
+    async def __aenter__(self) -> OpenSandboxClient:
+        headers = {}
+        if self.api_key_file is not None:
+            headers["OPEN-SANDBOX-API-KEY"] = self.api_key_file.read_text().strip()
+        self.session = ClientSession(headers=headers, 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(
+        self,
+        *,
+        pool: str,
+        timeout: int,
+        entrypoint: list[str],
+        env: dict[str, str],
+        metadata: dict[str, str],
+    ) -> dict[str, object]:
+        response = await self._request(
+            "POST",
+            "/v1/sandboxes",
+            json={
+                "timeout": timeout,
+                "entrypoint": entrypoint,
+                "env": env,
+                "metadata": metadata,
+                "extensions": {"poolRef": pool},
+            },
+        )
+        return await response.json()
+
+    async def get(self, sandbox_id: str) -> dict[str, object] | None:
+        try:
+            response = await self._request(
+                "GET", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
+            )
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def delete(self, sandbox_id: str) -> None:
+        try:
+            response = await self._request(
+                "DELETE", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
+            )
+            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("OpenSandboxClient is not open")
+        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)

src/gitea_dynamic_runner/opensandbox_identity.py

@@ -0,0 +1,183 @@
+"""Reconcile OpenSandbox Pod UIDs to narrowly scoped SPIRE entries."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import os
+from pathlib import Path
+
+from aiohttp import ClientResponseError
+
+from .sandbox_kubernetes import SandboxKubernetesClient, allocated_pod_name
+
+
+LOG = logging.getLogger(__name__)
+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("OPENSANDBOX_NAMESPACE", "opensandbox")
+SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
+SPIRE_CLUSTER_NAME = os.environ.get("SPIRE_CLUSTER_NAME", "sandbox-kata")
+SPIRE_CLASS_NAME = os.environ.get("SPIRE_CLASS_NAME", "spire-mgmt-spire")
+RUNNER_UID = int(os.environ.get("RUNNER_UID", "2000"))
+RECONCILE_INTERVAL = int(os.environ.get("RECONCILE_INTERVAL", "2"))
+
+
+def entry_name(sandbox_id: str) -> str:
+    suffix = hashlib.sha256(sandbox_id.encode()).hexdigest()[:12]
+    return f"gitea-ci-{suffix}"
+
+
+def sandbox_name(document: dict[str, object]) -> str | None:
+    metadata = document.get("metadata")
+    value = metadata.get("name") if isinstance(metadata, dict) else None
+    return value if isinstance(value, str) and value else None
+
+
+def is_runner_sandbox(document: dict[str, object]) -> bool:
+    metadata = document.get("metadata")
+    labels = metadata.get("labels") if isinstance(metadata, dict) else None
+    return isinstance(labels, dict) and labels.get("ci.ddupan.top/runner") == "true"
+
+
+def task_environment(document: dict[str, object]) -> dict[str, str]:
+    spec = document.get("spec")
+    task = spec.get("taskTemplate") if isinstance(spec, dict) else None
+    task_spec = task.get("spec") if isinstance(task, dict) else None
+    process = task_spec.get("process") if isinstance(task_spec, dict) else None
+    values = process.get("env") if isinstance(process, dict) else None
+    result: dict[str, str] = {}
+    if not isinstance(values, list):
+        return result
+    for item in values:
+        if not isinstance(item, dict):
+            continue
+        name, value = item.get("name"), item.get("value")
+        if isinstance(name, str) and isinstance(value, str):
+            result[name] = value
+    return result
+
+
+def identity_entry(
+    *, sandbox_id: str, pod_uid: str, spiffe_id: str
+) -> dict[str, object]:
+    expected_prefix = f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/"
+    if not spiffe_id.startswith(expected_prefix):
+        raise ValueError("runner SPIFFE ID is outside the CI namespace")
+    return {
+        "apiVersion": "spire.spiffe.io/v1alpha1",
+        "kind": "ClusterStaticEntry",
+        "metadata": {
+            "name": entry_name(sandbox_id),
+            "labels": {
+                "app.kubernetes.io/name": "gitea-dynamic-runner",
+                "app.kubernetes.io/component": "opensandbox-identity",
+                "ci.ddupan.top/sandbox-id": sandbox_id,
+            },
+        },
+        "spec": {
+            "className": SPIRE_CLASS_NAME,
+            "parentID": (
+                f"spiffe://{SPIFFE_TRUST_DOMAIN}/spire/agent/k8s_psat/"
+                f"{SPIRE_CLUSTER_NAME}/pod/{pod_uid}"
+            ),
+            "spiffeID": spiffe_id,
+            "selectors": [f"unix:uid:{RUNNER_UID}"],
+        },
+    }
+
+
+async def reconcile(client: SandboxKubernetesClient) -> None:
+    sandboxes = await client.list_batchsandboxes()
+    live_names = {
+        name
+        for document in sandboxes
+        if is_runner_sandbox(document) and (name := sandbox_name(document))
+    }
+    entries = await client.list_entries()
+    existing = {
+        name
+        for document in entries
+        if (name := sandbox_name(document)) is not None
+    }
+
+    for document in sandboxes:
+        if not is_runner_sandbox(document):
+            continue
+        name = sandbox_name(document)
+        pod_name = allocated_pod_name(document)
+        spiffe_id = task_environment(document).get("CI_SPIFFE_ID")
+        if not name or not pod_name or not spiffe_id or entry_name(name) in existing:
+            continue
+        pod = await client.get_pod(pod_name)
+        metadata = pod.get("metadata") if isinstance(pod, dict) else None
+        pod_uid = metadata.get("uid") if isinstance(metadata, dict) else None
+        if not isinstance(pod_uid, str) or not pod_uid:
+            continue
+        try:
+            await client.create_entry(
+                identity_entry(
+                    sandbox_id=name,
+                    pod_uid=pod_uid,
+                    spiffe_id=spiffe_id,
+                )
+            )
+        except ClientResponseError as error:
+            if error.status != 409:
+                raise
+        LOG.info(
+            "SPIRE entry ready sandbox=%s pod=%s pod_uid=%s spiffe_id=%s",
+            name,
+            pod_name,
+            pod_uid,
+            spiffe_id,
+        )
+
+    for document in entries:
+        metadata = document.get("metadata")
+        labels = metadata.get("labels") if isinstance(metadata, dict) else None
+        sandbox_id = labels.get("ci.ddupan.top/sandbox-id") if isinstance(labels, dict) else None
+        name = metadata.get("name") if isinstance(metadata, dict) else None
+        if (
+            isinstance(name, str)
+            and isinstance(sandbox_id, str)
+            and sandbox_id not in live_names
+        ):
+            await client.delete_entry(name)
+            LOG.info("removed stale SPIRE entry=%s sandbox=%s", name, sandbox_id)
+
+
+async def main() -> None:
+    async with SandboxKubernetesClient(
+        api_url=KUBERNETES_API,
+        token_file=KUBERNETES_TOKEN_FILE,
+        ca_file=KUBERNETES_CA_FILE,
+        namespace=NAMESPACE,
+    ) as client:
+        while True:
+            try:
+                await reconcile(client)
+            except Exception:
+                LOG.exception("OpenSandbox identity reconcile failed")
+            await asyncio.sleep(RECONCILE_INTERVAL)
+
+
+def cli() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    cli()

src/gitea_dynamic_runner/opensandbox_worker.py

@@ -0,0 +1,194 @@
+"""Direct OpenSandbox lifecycle scheduler used by the webhook controller."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import secrets
+import uuid
+from collections.abc import Callable
+
+from .models import RunnerRequest
+from .opensandbox import OpenSandboxClient
+from .pod_worker import identity_path
+
+
+LOG = logging.getLogger(__name__)
+SANDBOX_TIMEOUT = int(os.environ.get("RUNNER_SANDBOX_TIMEOUT", str(4 * 60 * 60)))
+SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
+TOKEN_BASE_URL = os.environ.get(
+    "RUNNER_TOKEN_BASE_URL", "http://192.168.10.127:8787/token"
+).rstrip("/")
+
+
+def sandbox_request(
+    request: RunnerRequest,
+    runner_name: str,
+    registration_token_url: str,
+) -> dict[str, object]:
+    path = identity_path(request.repository, request.job_name)
+    return {
+        "pool": f"ci-{request.backend}",
+        "timeout": SANDBOX_TIMEOUT,
+        "entrypoint": ["/usr/local/libexec/gitea-opensandbox-runner"],
+        "env": {
+            "GITEA_RUNNER_NAME": runner_name,
+            "GITEA_RUNNER_LABELS": f"self-hosted:host,{request.backend}:host",
+            "GITEA_RUNNER_REGISTRATION_TOKEN_URL": registration_token_url,
+            "GITEA_RUNNER_EPHEMERAL": "1",
+            "GITEA_RUNNER_ONCE": "1",
+            "CONFIG_FILE": "/etc/gitea-runner/config.yaml",
+            "CI_SPIFFE_ID": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}",
+            "SPIFFE_ENDPOINT_SOCKET": (
+                "unix:///run/spire/agent-sockets/spire-agent.sock"
+            ),
+        },
+        "metadata": {
+            "ci.ddupan.top/runner": "true",
+            "ci.ddupan.top/job-id": str(request.job_id),
+            "ci.ddupan.top/run-id": str(request.run_id),
+            "ci.ddupan.top/runner-name": runner_name,
+        },
+    }
+
+
+class RegistrationTokens:
+    """Single-use registration-token URLs; values never enter Sandbox CRs."""
+
+    def __init__(self, token: bytes) -> None:
+        self._token = token.strip()
+        self._values: dict[str, bytes] = {}
+        self._lock = asyncio.Lock()
+
+    async def issue(self) -> tuple[str, str]:
+        nonce = secrets.token_urlsafe(32)
+        async with self._lock:
+            self._values[nonce] = self._token
+        return nonce, f"{TOKEN_BASE_URL}/{nonce}"
+
+    async def consume(self, nonce: str) -> bytes | None:
+        async with self._lock:
+            return self._values.pop(nonce, None)
+
+    async def revoke(self, nonce: str) -> None:
+        async with self._lock:
+            self._values.pop(nonce, None)
+
+
+class OpenSandboxScheduler:
+    def __init__(
+        self,
+        client: OpenSandboxClient,
+        tokens: RegistrationTokens,
+        *,
+        on_finished: Callable[[int], None] | None = None,
+    ) -> None:
+        self.client = client
+        self.tokens = tokens
+        self.on_finished = on_finished
+        self.active: dict[int, asyncio.Task[None]] = {}
+        self.runners: dict[str, tuple[int, str, str]] = {}
+
+    async def create(self, request: RunnerRequest) -> str:
+        if request.job_id in self.active:
+            raise ValueError(f"job {request.job_id} already has an active sandbox")
+        runner_name = f"gitea-{request.backend}-{uuid.uuid4().hex[:12]}"
+        nonce, token_url = await self.tokens.issue()
+        try:
+            response = await self.client.create(
+                **sandbox_request(request, runner_name, token_url)
+            )
+            sandbox_id = response.get("id")
+            if not isinstance(sandbox_id, str) or not sandbox_id:
+                raise RuntimeError("OpenSandbox create response has no id")
+        except Exception:
+            await self.tokens.revoke(nonce)
+            raise
+
+        task = asyncio.create_task(
+            self._monitor(request, sandbox_id, nonce, runner_name),
+            name=f"opensandbox-{sandbox_id}",
+        )
+        task.add_done_callback(self._report)
+        self.active[request.job_id] = task
+        self.runners[runner_name] = (request.job_id, sandbox_id, nonce)
+        LOG.info(
+            "OpenSandbox runner created sandbox=%s runner=%s job_id=%d "
+            "repository=%s job_name=%r pool=ci-%s",
+            sandbox_id,
+            runner_name,
+            request.job_id,
+            request.repository,
+            request.job_name,
+            request.backend,
+        )
+        return sandbox_id
+
+    async def _monitor(
+        self,
+        request: RunnerRequest,
+        sandbox_id: str,
+        nonce: str,
+        runner_name: str,
+    ) -> None:
+        try:
+            deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT
+            while asyncio.get_running_loop().time() < deadline:
+                sandbox = await self.client.get(sandbox_id)
+                if sandbox is None:
+                    return
+                status = sandbox.get("status")
+                state = status.get("state") if isinstance(status, dict) else None
+                if state in {"Terminated", "Failed"}:
+                    return
+                await asyncio.sleep(2)
+            raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out")
+        finally:
+            await self._cleanup(request.job_id, sandbox_id, nonce, runner_name)
+
+    async def complete(self, runner_name: str) -> bool:
+        """Delete the sandbox which actually ran a completed Gitea job."""
+        state = self.runners.get(runner_name)
+        if state is None:
+            return False
+        job_id, sandbox_id, nonce = state
+        task = self.active.get(job_id)
+        if task is not None:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        await self._cleanup(job_id, sandbox_id, nonce, runner_name)
+        return True
+
+    async def _cleanup(
+        self,
+        job_id: int,
+        sandbox_id: str,
+        nonce: str,
+        runner_name: str,
+    ) -> None:
+        """Revoke and delete once, including cancellation-before-start races."""
+        if self.runners.pop(runner_name, None) is None:
+            return
+        await self.tokens.revoke(nonce)
+        try:
+            await self.client.delete(sandbox_id)
+        finally:
+            self.active.pop(job_id, None)
+            if self.on_finished is not None:
+                self.on_finished(job_id)
+
+    @staticmethod
+    def _report(task: asyncio.Task[None]) -> None:
+        if not task.cancelled() and (error := task.exception()) is not None:
+            LOG.error(
+                "OpenSandbox lifecycle task failed",
+                exc_info=(type(error), error, error.__traceback__),
+            )
+
+    async def close(self) -> None:
+        tasks = list(self.active.values())
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)

src/gitea_dynamic_runner/sandbox_kubernetes.py

@@ -0,0 +1,148 @@
+"""Kubernetes resources that bind an OpenSandbox Kata guest to SPIRE."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import ssl
+from urllib.parse import quote
+
+from aiohttp import ClientResponseError, ClientSession, TCPConnector
+
+
+class SandboxKubernetesClient:
+    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) -> SandboxKubernetesClient:
+        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 get_batchsandbox(self, sandbox_id: str) -> dict[str, object] | None:
+        try:
+            response = await self._request(
+                "GET", f"{self._batchsandboxes_path()}/{quote(sandbox_id, safe='')}"
+            )
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def list_batchsandboxes(self) -> list[dict[str, object]]:
+        response = await self._request("GET", self._batchsandboxes_path())
+        document = await response.json()
+        items = document.get("items") if isinstance(document, dict) else None
+        return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
+
+    async def get_pod(self, name: str) -> dict[str, object] | None:
+        try:
+            response = await self._request(
+                "GET", f"{self._pods_path()}/{quote(name, safe='')}"
+            )
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def create_entry(self, manifest: dict[str, object]) -> dict[str, object]:
+        response = await self._request(
+            "POST", self._entries_path(), json=manifest
+        )
+        return await response.json()
+
+    async def list_entries(self) -> list[dict[str, object]]:
+        response = await self._request(
+            "GET",
+            f"{self._entries_path()}?labelSelector="
+            "app.kubernetes.io%2Fcomponent%3Dopensandbox-identity",
+        )
+        document = await response.json()
+        items = document.get("items") if isinstance(document, dict) else None
+        return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
+
+    async def get_entry(self, name: str) -> dict[str, object] | None:
+        try:
+            response = await self._request(
+                "GET", f"{self._entries_path()}/{quote(name, safe='')}"
+            )
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def delete_entry(self, name: str) -> None:
+        try:
+            response = await self._request(
+                "DELETE",
+                f"{self._entries_path()}/{quote(name, safe='')}",
+                json={"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("SandboxKubernetesClient is not open")
+        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
+
+    def _batchsandboxes_path(self) -> str:
+        return (
+            "/apis/sandbox.opensandbox.io/v1alpha1/namespaces/"
+            f"{quote(self.namespace, safe='')}/batchsandboxes"
+        )
+
+    def _pods_path(self) -> str:
+        return f"/api/v1/namespaces/{quote(self.namespace, safe='')}/pods"
+
+    @staticmethod
+    def _entries_path() -> str:
+        return "/apis/spire.spiffe.io/v1alpha1/clusterstaticentries"
+
+
+def allocated_pod_name(batchsandbox: dict[str, object]) -> str | None:
+    metadata = batchsandbox.get("metadata")
+    if not isinstance(metadata, dict):
+        return None
+    annotations = metadata.get("annotations")
+    if not isinstance(annotations, dict):
+        return None
+    raw = annotations.get("sandbox.opensandbox.io/alloc-status")
+    if not isinstance(raw, str):
+        return None
+    try:
+        allocation = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(allocation, dict):
+        return None
+    pods = allocation.get("pods")
+    if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], str):
+        return None
+    return pods[0]

src/gitea_dynamic_runner/worker.py

@@ -31,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()
@@ -45,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:
@@ -162,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()

tests/test_controller.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import hmac
 import json
@@ -31,8 +32,14 @@ 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(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)
 
 
@@ -95,3 +102,35 @@ def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):
     assert controller.valid_signature(body, digest)
     assert controller.valid_signature(body, f"sha256={digest}")
     assert not controller.valid_signature(body, "bad")
+
+
+async def test_completed_lifecycle_cancellation_acks_message():
+    request = RunnerRequest.from_webhook(queued_job())
+    assert request is not None
+
+    class Message:
+        data = request.to_json()
+        acked = False
+
+        async def ack(self):
+            self.acked = True
+
+        async def nak(self, **kwargs):
+            raise AssertionError(f"unexpected NAK: {kwargs}")
+
+        async def in_progress(self):
+            pass
+
+    class Scheduler:
+        active = {}
+
+        async def create(self, runner_request):
+            async def completed():
+                self.active.pop(runner_request.job_id, None)
+                raise asyncio.CancelledError
+
+            self.active[runner_request.job_id] = asyncio.create_task(completed())
+
+    message = Message()
+    await controller.run_message(message, Scheduler())
+    assert message.acked is True

tests/test_opensandbox_identity.py

@@ -0,0 +1,53 @@
+import json
+
+from gitea_dynamic_runner.opensandbox_identity import (
+    identity_entry,
+    is_runner_sandbox,
+    task_environment,
+)
+from gitea_dynamic_runner.sandbox_kubernetes import allocated_pod_name
+
+
+def batchsandbox():
+    return {
+        "metadata": {
+            "name": "sandbox-1",
+            "labels": {"ci.ddupan.top/runner": "true"},
+            "annotations": {
+                "sandbox.opensandbox.io/alloc-status": json.dumps(
+                    {"pods": ["ci-vm-pod-1"]}
+                )
+            },
+        },
+        "spec": {
+            "taskTemplate": {
+                "spec": {
+                    "process": {
+                        "env": [
+                            {
+                                "name": "CI_SPIFFE_ID",
+                                "value": "spiffe://ddupan.top/ci/org/repo/test",
+                            }
+                        ]
+                    }
+                }
+            }
+        },
+    }
+
+
+def test_extracts_allocation_and_identity_environment():
+    document = batchsandbox()
+    assert is_runner_sandbox(document)
+    assert allocated_pod_name(document) == "ci-vm-pod-1"
+    assert task_environment(document)["CI_SPIFFE_ID"].endswith("/org/repo/test")
+
+
+def test_entry_binds_exact_pod_agent_and_runner_uid():
+    entry = identity_entry(
+        sandbox_id="sandbox-1",
+        pod_uid="pod-uid",
+        spiffe_id="spiffe://ddupan.top/ci/org/repo/test",
+    )
+    assert entry["spec"]["parentID"].endswith("/sandbox-kata/pod/pod-uid")
+    assert entry["spec"]["selectors"] == ["unix:uid:2000"]

tests/test_opensandbox_worker.py

@@ -0,0 +1,101 @@
+import asyncio
+
+import pytest
+
+from gitea_dynamic_runner.models import RunnerRequest
+from gitea_dynamic_runner.opensandbox_worker import (
+    OpenSandboxScheduler,
+    RegistrationTokens,
+    sandbox_request,
+)
+
+
+def request(backend: str = "vm") -> RunnerRequest:
+    return RunnerRequest(
+        job_id=42,
+        run_id=7,
+        backend=backend,
+        repository="panxiao81/example",
+        job_name="publish image",
+        labels=("self-hosted", backend),
+    )
+
+
+def test_sandbox_request_uses_pool_identity_and_one_time_token_url():
+    document = sandbox_request(
+        request(), "gitea-vm-abcd", "http://scheduler/token/nonce"
+    )
+    assert document["pool"] == "ci-vm"
+    assert document["entrypoint"] == ["/usr/local/libexec/gitea-opensandbox-runner"]
+    assert document["env"]["GITEA_RUNNER_LABELS"] == "self-hosted:host,vm:host"
+    assert document["env"]["GITEA_RUNNER_REGISTRATION_TOKEN_URL"].endswith("/nonce")
+    assert document["env"]["CI_SPIFFE_ID"] == (
+        "spiffe://ddupan.top/ci/panxiao81/example/publish-image-3e72cdc4a97e"
+    )
+    assert document["metadata"]["ci.ddupan.top/runner"] == "true"
+
+
+async def test_registration_token_is_single_use(monkeypatch):
+    monkeypatch.setattr(
+        "gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
+        "http://scheduler/token",
+    )
+    tokens = RegistrationTokens(b"secret\n")
+    nonce, url = await tokens.issue()
+    assert url == f"http://scheduler/token/{nonce}"
+    assert await tokens.consume(nonce) == b"secret"
+    assert await tokens.consume(nonce) is None
+
+
+class FakeOpenSandbox:
+    def __init__(self):
+        self.created = None
+        self.deleted = []
+
+    async def create(self, **document):
+        self.created = document
+        return {"id": "sandbox-1"}
+
+    async def get(self, sandbox_id):
+        return {"status": {"state": "Terminated"}}
+
+    async def delete(self, sandbox_id):
+        self.deleted.append(sandbox_id)
+
+
+async def test_scheduler_creates_monitors_and_deletes(monkeypatch):
+    monkeypatch.setattr(
+        "gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
+        "http://scheduler/token",
+    )
+    client = FakeOpenSandbox()
+    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
+    assert await scheduler.create(request()) == "sandbox-1"
+    await scheduler.active[42]
+    assert client.created["pool"] == "ci-vm"
+    assert client.deleted == ["sandbox-1"]
+
+
+async def test_scheduler_rejects_duplicate_active_job():
+    client = FakeOpenSandbox()
+    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
+    scheduler.active[42] = asyncio.Future()
+    with pytest.raises(ValueError, match="already"):
+        await scheduler.create(request())
+
+
+async def test_scheduler_deletes_sandbox_for_completed_runner():
+    class RunningOpenSandbox(FakeOpenSandbox):
+        async def get(self, sandbox_id):
+            return {"status": {"state": "Running"}}
+
+    client = RunningOpenSandbox()
+    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
+    await scheduler.create(request())
+    runner_name = next(iter(scheduler.runners))
+
+    assert await scheduler.complete(runner_name) is True
+    assert client.deleted == ["sandbox-1"]
+    assert scheduler.active == {}
+    assert scheduler.runners == {}
+    assert await scheduler.complete(runner_name) is False