ci: 为 Kata nested kubelet 补充 kmsg 设备

.gitea/workflows/test.yml

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

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

@@ -1,31 +1,154 @@
+---
 name: VM kind smoke
 
+# yamllint disable-line rule:truthy
 on:
   workflow_dispatch:
 
 jobs:
   kind:
-    runs-on: [self-hosted, vm]
+    runs-on: [self-hosted, vm-dev]
     steps:
+      - name: Start job-local Docker
+        shell: bash
+        run: |
+          set -euo pipefail
+          if [[ ! -e /dev/kmsg ]]; then
+            sudo mknod /dev/kmsg c 1 11
+          fi
+          sudo install -d /var/lib/docker
+          sudo truncate -s 20G /tmp/docker-data.img
+          sudo mkfs.ext4 -F /tmp/docker-data.img
+          if [[ ! -e /dev/loop-control ]]; then
+            sudo mknod /dev/loop-control c 10 237
+          fi
+          for minor in {0..7}; do
+            if [[ ! -e "/dev/loop${minor}" ]]; then
+              sudo mknod "/dev/loop${minor}" b 7 "$minor"
+            fi
+          done
+          loop_device=$(sudo losetup --find --show /tmp/docker-data.img)
+          sudo mount "$loop_device" /var/lib/docker
+          findmnt /var/lib/docker
+
+          # A nested systemd needs a domain cgroup namespace. This is the
+          # cgroup v2 nesting initialization performed by the official DinD
+          # entrypoint, kept here because the shared runner image deliberately
+          # does not carry a second DinD-specific entrypoint.
+          if [[ -f /sys/fs/cgroup/cgroup.controllers ]]; then
+            sudo sh -c '
+              mkdir -p /sys/fs/cgroup/init
+              while read -r pid; do
+                printf "%s\n" "$pid" \
+                  >/sys/fs/cgroup/init/cgroup.procs 2>/dev/null || true
+              done </sys/fs/cgroup/cgroup.procs
+              sed -e "s/ / +/g" -e "s/^/+/" \
+                /sys/fs/cgroup/cgroup.controllers \
+                >/sys/fs/cgroup/cgroup.subtree_control
+            '
+          fi
+
+          sudo nohup dockerd \
+            --host=unix:///var/run/docker.sock \
+            --storage-driver=overlay2 \
+            >/tmp/dockerd.log 2>&1 &
+          for _ in {1..60}; do
+            if docker info >/dev/null 2>&1; then
+              exit 0
+            fi
+            sleep 1
+          done
+          cat /tmp/dockerd.log
+          exit 1
+
       - name: Verify Docker
-        run: docker info
+        shell: bash
+        run: |
+          set -euo pipefail
+          docker info
+          echo '### runner cgroup'
+          cat /proc/self/cgroup
+          cat /sys/fs/cgroup/cgroup.type
+          cat /sys/fs/cgroup/cgroup.controllers
+          cat /sys/fs/cgroup/cgroup.subtree_control
+          echo '### nested private cgroup namespace'
+          docker run --rm --privileged --cgroupns=private alpine:3.22 \
+            sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
+          echo '### nested host cgroup namespace'
+          docker run --rm --privileged --cgroupns=host alpine:3.22 \
+            sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
 
       - name: Install kind
         shell: bash
         run: |
           set -euo pipefail
           version=v0.33.0
+          base_url="https://kind.sigs.k8s.io/dl/${version}"
           curl --fail --location --silent --show-error \
-            --output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
+            --output /tmp/kind "${base_url}/kind-linux-amd64"
           curl --fail --location --silent --show-error \
-            --output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
+            --output /tmp/kind.sha256sum \
-          printf '%s  %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
+            "${base_url}/kind-linux-amd64.sha256sum"
+          checksum=$(cut -d ' ' -f1 /tmp/kind.sha256sum)
+          printf '%s  %s\n' "$checksum" /tmp/kind | sha256sum --check
           chmod 0755 /tmp/kind
 
       - name: Create and delete kind cluster
         shell: bash
         run: |
           set -euo pipefail
-          trap '/tmp/kind delete cluster --name smoke' EXIT
+          diagnose_and_cleanup() {
-          /tmp/kind create cluster --name smoke --wait 180s
+            status=$?
+            node=smoke-control-plane
+            if (( status != 0 )) && docker inspect "$node" >/dev/null 2>&1; then
+              echo '::group::kind node inspect'
+              docker inspect "$node"
+              echo '::endgroup::'
+              echo '::group::kind node logs'
+              docker logs "$node" 2>&1 || true
+              echo '::endgroup::'
+              echo '::group::kind node guest state'
+              docker exec "$node" bash -c '
+                set +e
+                echo "### pid 1"
+                ps -p 1 -o pid,ppid,user,stat,comm,args
+                cat /proc/1/status
+                echo "### cgroup"
+                cat /proc/1/cgroup
+                findmnt -R /sys/fs/cgroup
+                stat -fc "%T %a" /sys/fs/cgroup
+                echo "### systemd"
+                systemctl --no-pager --failed
+                systemctl --no-pager status \
+                  multi-user.target containerd.service kubelet.service
+                echo "### CRI containers"
+                endpoint=unix:///run/containerd/containerd.sock
+                crictl --runtime-endpoint "$endpoint" ps --all
+                for id in $(
+                  crictl --runtime-endpoint "$endpoint" ps --all --quiet
+                ); do
+                  echo "### CRI container $id"
+                  crictl --runtime-endpoint "$endpoint" inspect "$id"
+                  crictl --runtime-endpoint "$endpoint" logs "$id"
+                done
+                echo "### containerd metadata"
+                timeout 10 ctr --namespace k8s.io containers list
+                timeout 10 ctr --namespace k8s.io snapshots list
+                echo "### runtime process stacks"
+                ps -e -o pid,ppid,stat,wchan:32,comm,args
+                for pid in $(pidof containerd containerd-shim-runc-v2); do
+                  echo "### kernel stack $pid"
+                  cat "/proc/$pid/stack"
+                done
+                kill -USR1 "$(pidof containerd)"
+                sleep 2
+                journalctl --no-pager -b -n 500
+              ' 2>&1 || true
+              echo '::endgroup::'
+            fi
+            /tmp/kind delete cluster --name smoke || true
+            exit "$status"
+          }
+          trap diagnose_and_cleanup EXIT
+          /tmp/kind create cluster --name smoke --wait 180s --retain
           /tmp/kind get clusters | grep -Fx smoke

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

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

README.md

@@ -15,6 +15,10 @@ runs-on: [self-hosted, vm]
 只执行一个 job，并在 job 结束后连同本地状态一起销毁。完整的设计约束见
 [`docs/design-principles.md`](docs/design-principles.md)。
 
+集成期间可将 `VM_RUNNER_LABEL=vm-dev`，只接取显式使用
+`runs-on: [self-hosted, vm-dev]` 的测试任务；生产 `vm` job 将保持在 Gitea pending，
+不会在 backend 修复过程中继续涌入。
+
 目标 Go controller 组件：
 
 - `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment

cmd/gitea-dynamic-runner/controller.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
+	"log/slog"
 	"net/http"
 	"os"
 	"slices"
@@ -49,7 +49,7 @@ type controllerConfig struct {
 	PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
 	SPIREAgentID                                                        string
 	PodExecutorUID, PodCapacity                                         int
-	OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool                  string
+	OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool, VMRunnerLabel   string
 	VMTimeout, VMCapacity                                               int
 }
 
@@ -131,7 +131,7 @@ func runController(ctx context.Context) error {
 		labels = append(labels, string(taskassignment.BackendPod))
 	}
 	if slices.Contains(config.Components, controller.VMWorker) {
-		labels = append(labels, string(taskassignment.BackendVM))
+		labels = append(labels, config.VMRunnerLabel)
 	}
 	poller := taskscheduler.Poller{
 		Client: giteaClient,
@@ -139,7 +139,7 @@ func runController(ctx context.Context) error {
 			JetStream: producerJS, SubjectBase: config.SubjectBase,
 		}},
 		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
-		OnError: func(err error) { log.Printf("scheduler: %v", err) },
+		OnError: func(err error) { slog.Error("scheduler error", "component", "scheduler", "error", err) },
 	}
 	kubernetesConfig, err := rest.InClusterConfig()
 	if err != nil {
@@ -190,11 +190,13 @@ func runController(ctx context.Context) error {
 				return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
 			}
 		}
-		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendPod)}, registry, podPool)
 		if err != nil {
 			return err
 		}
-		lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
+		lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) {
+			slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendPod, "error", err)
+		}}
 		components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return component.Run(groupContext) })
@@ -220,11 +222,13 @@ func runController(ctx context.Context) error {
 				return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
 			}
 		}
-		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendVM)}, registry, vmPool)
 		if err != nil {
 			return err
 		}
-		lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
+		lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) {
+			slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendVM, "error", err)
+		}}
 		components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return component.Run(groupContext) })
@@ -294,11 +298,21 @@ func workerComponent(ctx context.Context, js jetstream.JetStream, config control
 	}
 	return assignmentqueue.ConsumerComponent{
 		Consumer: consumer, Capacity: capacity,
-		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
+		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission, OnEvent: func(event assignmentqueue.Event) {
-		OnError:   func(err error) { log.Printf("%s worker: %v", backend, err) },
+			slog.Info("assignment transition", "component", "worker", "event", event.Name, "backend", event.Backend, "assignment", event.AssignmentID, "retry_delay", event.RetryDelay)
+		}},
+		OnError: func(err error) {
+			slog.Error("assignment processing error", "component", "worker", "backend", backend, "error", err)
+		},
 	}, nil
 }
 
+func workerEventLogger(backend taskassignment.Backend) func(taskworker.Event) {
+	return func(event taskworker.Event) {
+		slog.Info("executor transition", "component", "worker", "event", event.Name, "backend", backend, "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase)
+	}
+}
+
 func loadControllerConfig() (controllerConfig, error) {
 	selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
 	if err != nil {
@@ -345,7 +359,7 @@ func loadControllerConfig() (controllerConfig, error) {
 		FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
 		PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),
 		SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), SPIREAgentID: os.Getenv("SPIRE_AGENT_ID"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4),
-		OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
+		OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMRunnerLabel: env("VM_RUNNER_LABEL", "vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
 	}
 	if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
 		return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required")
@@ -357,6 +371,9 @@ func loadControllerConfig() (controllerConfig, error) {
 		return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
 	}
 	if slices.Contains(selection, controller.VMWorker) {
+		if config.VMRunnerLabel != "vm" && config.VMRunnerLabel != "vm-dev" {
+			return controllerConfig{}, errors.New("VM_RUNNER_LABEL must be vm or vm-dev")
+		}
 		if config.OpenSandboxURL == "" {
 			return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
 		}

cmd/gitea-dynamic-runner/controller_test.go

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

cmd/gitea-dynamic-runner/main.go

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

docs/opensandbox-runner.md

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

internal/assignmentqueue/jetstream.go

@@ -78,6 +78,22 @@ type Processor struct {
 	Admission    Admission
 	RetryDelay   time.Duration
 	ClaimTimeout time.Duration
+	OnEvent      func(Event)
+}
+
+// Event describes a non-sensitive assignment handoff transition. It never
+// contains task payloads, credentials, capabilities, or workload identities.
+type Event struct {
+	Name         string
+	AssignmentID string
+	Backend      taskassignment.Backend
+	RetryDelay   time.Duration
+}
+
+func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) {
+	if p.OnEvent != nil {
+		p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay})
+	}
 }
 
 func (p Processor) Process(ctx context.Context, message Message) error {
@@ -88,6 +104,7 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 	if err != nil {
 		return errors.Join(err, message.TermWithReason("invalid assignment"))
 	}
+	p.event("received", assignment, 0)
 	if _, err := p.Claims.Offer(assignment); err != nil {
 		return errors.Join(err, message.TermWithReason("conflicting assignment"))
 	}
@@ -96,8 +113,10 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 		if delay <= 0 {
 			delay = 2 * time.Second
 		}
+		p.event("capacity_wait", assignment, delay)
 		return message.NakWithDelay(delay)
 	}
+	p.event("capacity_acquired", assignment, 0)
 	accepted, err := p.Accepter.Accept(ctx, assignment)
 	if err != nil {
 		p.Admission.Release(assignment.ID)
@@ -105,9 +124,11 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 		if delay <= 0 {
 			delay = 15 * time.Second
 		}
+		p.event("backend_retry", assignment, delay)
 		return errors.Join(err, message.NakWithDelay(delay))
 	}
 	if accepted {
+		p.event("backend_ready", assignment, 0)
 		timeout := p.ClaimTimeout
 		if timeout <= 0 {
 			timeout = 4 * time.Minute
@@ -120,17 +141,21 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 			if delay <= 0 {
 				delay = 2 * time.Second
 			}
+			p.event("claim_timeout", assignment, delay)
 			return errors.Join(err, message.NakWithDelay(delay))
 		}
+		p.event("runner_claimed", assignment, 0)
 		if err := message.DoubleAck(ctx); err != nil {
 			return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
 		}
+		p.event("acked", assignment, 0)
 		return nil
 	}
 	delay := p.RetryDelay
 	if delay <= 0 {
 		delay = 2 * time.Second
 	}
+	p.event("backend_pending", assignment, delay)
 	return message.NakWithDelay(delay)
 }
 
@@ -167,7 +192,7 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB
 		AckPolicy:     jetstream.AckExplicitPolicy,
 		AckWait:       5 * time.Minute,
 		MaxAckPending: capacity,
-		MaxDeliver:    20,
+		MaxDeliver:    1000,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)

internal/assignmentqueue/jetstream_test.go

@@ -126,13 +126,23 @@ func encodedAssignment(t *testing.T) []byte {
 
 func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
 	message := &fakeMessage{data: encodedAssignment(t)}
-	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
+	var events []Event
+	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}, OnEvent: func(event Event) { events = append(events, event) }}
 	if err := processor.Process(context.Background(), message); err != nil {
 		t.Fatal(err)
 	}
 	if message.acked != 1 || message.nacked != 0 {
 		t.Fatalf("message = %#v", message)
 	}
+	want := []string{"received", "capacity_acquired", "backend_ready", "runner_claimed", "acked"}
+	if len(events) != len(want) {
+		t.Fatalf("events = %#v", events)
+	}
+	for index := range want {
+		if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod {
+			t.Fatalf("event[%d] = %#v", index, events[index])
+		}
+	}
 }
 
 func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
@@ -205,7 +215,7 @@ func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
 	if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
 		t.Fatal(err)
 	}
-	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 {
+	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
 		t.Fatalf("config = %#v", manager.config)
 	}
 }

internal/taskassignment/assignment.go

@@ -102,7 +102,7 @@ func backendFromTask(task *runnerv1.Task) (Backend, error) {
 		return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
 	}
 	hasPod := slices.Contains(labels, string(BackendPod))
-	hasVM := slices.Contains(labels, string(BackendVM))
+	hasVM := slices.Contains(labels, string(BackendVM)) || slices.Contains(labels, "vm-dev")
 	if hasPod == hasVM {
 		return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
 	}

internal/taskassignment/assignment_test.go

@@ -28,6 +28,7 @@ func TestNewSelectsBackendFromRunsOn(t *testing.T) {
 	}{
 		{"[self-hosted, pod]", BackendPod},
 		{"[self-hosted, vm]", BackendVM},
+		{"[self-hosted, vm-dev]", BackendVM},
 	} {
 		assignment, err := New(task(t, test.labels), "ddupan.top")
 		if err != nil {

internal/taskworker/worker.go

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

internal/taskworker/worker_test.go

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