重构为直接 OpenSandbox 生命周期调度

src/gitea_dynamic_runner/controller.py

@@ -1,63 +1,54 @@
 #!/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 hashlib
 import hmac
 import json
 import logging
 import os
-import ssl
 from pathlib import Path
 
-import nats
 from aiohttp import web
-from nats.js.api import 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")
-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"))
+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",
+    )
+)
 
 
 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", "")):
@@ -70,49 +61,28 @@ async def webhook(request: web.Request) -> web.Response:
     LOG.info("workflow_job webhook received %s", context)
 
     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")
+    scheduler: OpenSandboxScheduler = request.app["scheduler"]
+    try:
+        sandbox_id = await scheduler.create(runner_request)
+    except ValueError as error:
+        LOG.info("duplicate workflow_job webhook ignored: %s", error)
+        return web.Response(status=202, text="already scheduled\n")
+    except Exception:
+        LOG.exception("failed to create OpenSandbox runner %s", context)
+        raise web.HTTPServiceUnavailable(text="sandbox unavailable\n")
+    return web.Response(status=202, text=f"sandbox={sandbox_id}\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 +101,42 @@ 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()
-    await ensure_stream(app["js"])
-    yield
-    await nc.drain()
+
+
+async def opensandbox_context(app: web.Application):
+    tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
+    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
+        yield
+        await scheduler.close()
 
 
 def create_app() -> web.Application:
     app = web.Application(client_max_size=1024 * 1024)
-    app.cleanup_ctx.append(nats_context)
+    app.cleanup_ctx.append(opensandbox_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_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

@@ -1,68 +1,51 @@
-#!/usr/bin/env python3
-"""JetStream worker that provisions disposable OpenSandbox Kata runners."""
+"""Direct OpenSandbox lifecycle scheduler used by the webhook controller."""
 
 from __future__ import annotations
 
 import asyncio
-import hashlib
-import json
 import logging
 import os
-from pathlib import Path
-import ssl
+import secrets
 import uuid
-
-import nats
-from nats.errors import TimeoutError
-from nats.js.api import AckPolicy, ConsumerConfig
+from collections.abc import Callable
 
 from .models import RunnerRequest
 from .opensandbox import OpenSandboxClient
-from .pod_worker import heartbeat, identity_path
-from .sandbox_kubernetes import SandboxKubernetesClient, allocated_pod_name
+from .pod_worker import identity_path
 
 
 LOG = logging.getLogger(__name__)
-STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
-REQUEST_SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm")
-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("OPENSANDBOX_NAMESPACE", "opensandbox")
-OPENSANDBOX_API = os.environ.get(
-    "OPENSANDBOX_API", "http://opensandbox-server.opensandbox-system.svc"
-)
-OPENSANDBOX_API_KEY_FILE_VALUE = os.environ.get("OPENSANDBOX_API_KEY_FILE")
-OPENSANDBOX_API_KEY_FILE = Path(OPENSANDBOX_API_KEY_FILE_VALUE) if OPENSANDBOX_API_KEY_FILE_VALUE else None
-OPENSANDBOX_POOL = os.environ.get("OPENSANDBOX_POOL", "ci-vm")
 SANDBOX_TIMEOUT = int(os.environ.get("RUNNER_SANDBOX_TIMEOUT", str(4 * 60 * 60)))
-ALLOCATION_TIMEOUT = int(os.environ.get("RUNNER_ALLOCATION_TIMEOUT", "120"))
-CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "2"))
 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"))
+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) -> dict[str, object]:
-    if request.backend != "vm":
-        raise ValueError("OpenSandbox backend only accepts vm requests")
+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": OPENSANDBOX_POOL,
+        "pool": f"ci-{request.backend}",
         "timeout": SANDBOX_TIMEOUT,
         "entrypoint": ["/usr/local/libexec/gitea-opensandbox-runner"],
         "env": {
             "GITEA_RUNNER_NAME": runner_name,
-            "GITEA_RUNNER_LABELS": "self-hosted:host,vm:host",
+            "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",
+            "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,
@@ -70,207 +53,111 @@ def sandbox_request(request: RunnerRequest, runner_name: str) -> dict[str, objec
     }
 
 
-def entry_name(sandbox_id: str) -> str:
-    suffix = hashlib.sha256(sandbox_id.encode()).hexdigest()[:12]
-    return f"gitea-ci-{suffix}"
+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)
 
 
-def identity_entry(
-    request: RunnerRequest,
-    *,
-    sandbox_id: str,
-    pod_uid: str,
-) -> dict[str, object]:
-    path = identity_path(request.repository, request.job_name)
-    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": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}",
-            "selectors": [f"unix:uid:{RUNNER_UID}"],
-        },
-    }
+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]] = {}
 
-
-async def wait_for_allocation(
-    client: SandboxKubernetesClient, sandbox_id: str
-) -> tuple[str, str]:
-    deadline = asyncio.get_running_loop().time() + ALLOCATION_TIMEOUT
-    while asyncio.get_running_loop().time() < deadline:
-        batchsandbox = await client.get_batchsandbox(sandbox_id)
-        if batchsandbox is not None and (pod_name := allocated_pod_name(batchsandbox)):
-            pod = await client.get_pod(pod_name)
-            if pod is not None:
-                metadata = pod.get("metadata")
-                pod_uid = metadata.get("uid") if isinstance(metadata, dict) else None
-                if isinstance(pod_uid, str) and pod_uid:
-                    return pod_name, pod_uid
-        await asyncio.sleep(1)
-    raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} allocation timed out")
-
-
-async def wait_for_sandbox(client: OpenSandboxClient, sandbox_id: str) -> bool:
-    deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT
-    while asyncio.get_running_loop().time() < deadline:
-        sandbox = await client.get(sandbox_id)
-        if sandbox is None:
-            raise RuntimeError(f"OpenSandbox {sandbox_id} disappeared")
-        status = sandbox.get("status")
-        state = status.get("state") if isinstance(status, dict) else None
-        if state == "Terminated":
-            return True
-        if state == "Failed":
-            return False
-        await asyncio.sleep(2)
-    raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out")
-
-
-async def run_request(
-    message: object,
-    sandbox_client: OpenSandboxClient,
-    kubernetes_client: SandboxKubernetesClient,
-) -> None:
-    try:
-        request = RunnerRequest.from_json(message.data)
-    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
-        LOG.error("discarding invalid OpenSandbox request: %s", error)
-        await message.ack()
-        return
-    if request.backend != "vm":
-        LOG.error("discarding %s request received by OpenSandbox worker", request.backend)
-        await message.ack()
-        return
-
-    runner_name = f"gitea-vm-{uuid.uuid4().hex[:12]}"
-    sandbox_id: str | None = None
-    static_entry: str | None = None
-    stop = asyncio.Event()
-    pulse = asyncio.create_task(heartbeat(message, stop))
-    try:
-        create = await sandbox_client.create(**sandbox_request(request, runner_name))
-        candidate = create.get("id")
-        if not isinstance(candidate, str) or not candidate:
-            raise RuntimeError("OpenSandbox create response has no id")
-        sandbox_id = candidate
-        pod_name, pod_uid = await wait_for_allocation(kubernetes_client, sandbox_id)
-        manifest = identity_entry(request, sandbox_id=sandbox_id, pod_uid=pod_uid)
-        static_entry = entry_name(sandbox_id)
-        await kubernetes_client.create_entry(manifest)
-        LOG.info(
-            "OpenSandbox identity created sandbox=%s pod=%s pod_uid=%s entry=%s runner=%s job_id=%d",
-            sandbox_id, pod_name, pod_uid, static_entry, runner_name, request.job_id,
-        )
-        if await wait_for_sandbox(sandbox_client, sandbox_id):
-            await message.ack()
-        else:
-            await message.nak(delay=30)
-    except Exception:
-        LOG.exception("OpenSandbox runner failed sandbox=%s job_id=%d", sandbox_id, request.job_id)
-        await message.nak(delay=30)
-        raise
-    finally:
-        stop.set()
-        await pulse
+    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:
-            if static_entry is not None:
-                await kubernetes_client.delete_entry(static_entry)
-        finally:
-            if sandbox_id is not None:
-                await sandbox_client.delete(sandbox_id)
-
-
-async def consume_requests(
-    subscription: object,
-    sandbox_client: OpenSandboxClient,
-    kubernetes_client: SandboxKubernetesClient,
-) -> 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, sandbox_client, kubernetes_client)
+            response = await self.client.create(
+                **sandbox_request(request, runner_name, token_url)
             )
-            task.add_done_callback(_report_task)
-            active.add(task)
+            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
 
-
-def _report_task(task: asyncio.Task[None]) -> None:
-    if not task.cancelled() and (error := task.exception()) is not None:
-        LOG.error(
-            "OpenSandbox runner task failed",
-            exc_info=(type(error), error, error.__traceback__),
+        task = asyncio.create_task(
+            self._monitor(request, sandbox_id, nonce),
+            name=f"opensandbox-{sandbox_id}",
         )
+        task.add_done_callback(self._report)
+        self.active[request.job_id] = task
+        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 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="opensandbox-runner-worker",
-    )
-    js = nc.jetstream()
-    subscription = await js.pull_subscribe(
-        REQUEST_SUBJECT,
-        durable="vm",
-        stream=STREAM,
-        config=ConsumerConfig(
-            durable_name="vm",
-            filter_subject=REQUEST_SUBJECT,
-            ack_policy=AckPolicy.EXPLICIT,
-            ack_wait=5 * 60,
-            max_ack_pending=max(CAPACITY, 1),
-            max_deliver=5,
-        ),
-    )
-    async with (
-        OpenSandboxClient(
-            api_url=OPENSANDBOX_API,
-            api_key_file=OPENSANDBOX_API_KEY_FILE,
-        ) as sandbox_client,
-        SandboxKubernetesClient(
-            api_url=KUBERNETES_API,
-            token_file=KUBERNETES_TOKEN_FILE,
-            ca_file=KUBERNETES_CA_FILE,
-            namespace=NAMESPACE,
-        ) as kubernetes_client,
-    ):
+    async def _monitor(
+        self, request: RunnerRequest, sandbox_id: str, nonce: str
+    ) -> None:
         try:
-            await consume_requests(subscription, sandbox_client, kubernetes_client)
+            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 nc.drain()
+            await self.tokens.revoke(nonce)
+            try:
+                await self.client.delete(sandbox_id)
+            finally:
+                self.active.pop(request.job_id, None)
+                if self.on_finished is not None:
+                    self.on_finished(request.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__),
+            )
 
-def cli() -> None:
-    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
-    asyncio.run(main())
-
-
-if __name__ == "__main__":
-    cli()
+    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

@@ -51,6 +51,12 @@ class SandboxKubernetesClient:
             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(
@@ -68,6 +74,16 @@ class SandboxKubernetesClient:
         )
         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(