"""Kubernetes-backed worker backend for the primary MindRoom runtime."""

from __future__ import annotations

import logging
import threading
import time
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING

from mindroom.credential_policy import credential_service_policy
from mindroom.credentials import get_runtime_credentials_manager, sync_shared_credentials_to_worker
from mindroom.runtime_env_policy import CREDENTIALS_ENCRYPTION_KEY_ENV, credentials_encryption_key_value
from mindroom.tool_system.worker_routing import resolved_worker_key_scope, worker_dir_name, worker_id_for_key
from mindroom.workers.backend import (
    WorkerBackendError,
    effective_idle_status,
    filter_and_sort_worker_handles,
)
from mindroom.workers.backends._lifecycle import mark_worker_failed, mark_worker_idle, touch_worker_lifecycle
from mindroom.workers.models import (
    ProgressSink,
    WorkerHandle,
    WorkerMaintenanceResult,
    WorkerReadyPhase,
    WorkerReadyProgress,
    WorkerSpec,
    WorkerStatus,
)

from . import kubernetes_resources as resources
from .kubernetes_config import (
    KubernetesWorkerBackendConfig,
    credentials_encryption_key_hash,
    kubernetes_backend_config_signature,
)

if TYPE_CHECKING:
    from collections.abc import Callable
    from pathlib import Path

    from mindroom.constants import RuntimePaths

__all__ = [
    "KubernetesWorkerBackend",
    "KubernetesWorkerBackendConfig",
    "kubernetes_backend_config_signature",
]

_COLD_START_GRACE_SECONDS = 1.5
_WAITING_PROGRESS_INTERVAL_SECONDS = 5.0
_PROGRESS_REPORTER_JOIN_TIMEOUT_SECONDS = 1.0
_READY_WORKER_REVALIDATE_SECONDS = 300.0

logger = logging.getLogger(__name__)


@dataclass
class _ProgressReporterState:
    started_at: float
    cold_start_emitted: bool = False
    next_waiting_elapsed: float = _WAITING_PROGRESS_INTERVAL_SECONDS
    reporter_done: bool = False


@dataclass(frozen=True, slots=True)
class _ReadyWorkerCacheEntry:
    spec: WorkerSpec
    handle: WorkerHandle
    validated_at: float
    credentials_encryption_key_hash: str | None


def _noop_finalize_progress(_phase: WorkerReadyPhase, _error: str | None) -> None:
    del _phase, _error


def _progress_event(
    *,
    phase: WorkerReadyPhase,
    worker_key: str,
    backend_name: str,
    elapsed_seconds: float,
    error: str | None = None,
) -> WorkerReadyProgress:
    return WorkerReadyProgress(
        phase=phase,
        worker_key=worker_key,
        backend_name=backend_name,
        elapsed_seconds=elapsed_seconds,
        error=error,
    )


def _pending_progress_events(
    *,
    state: _ProgressReporterState,
    worker_key: str,
    backend_name: str,
) -> list[WorkerReadyProgress]:
    elapsed_seconds = max(0.0, time.monotonic() - state.started_at)
    events: list[WorkerReadyProgress] = []
    if not state.cold_start_emitted and elapsed_seconds >= _COLD_START_GRACE_SECONDS:
        events.append(
            _progress_event(
                phase="cold_start",
                worker_key=worker_key,
                backend_name=backend_name,
                elapsed_seconds=elapsed_seconds,
            ),
        )
        state.cold_start_emitted = True
    while state.cold_start_emitted and elapsed_seconds >= state.next_waiting_elapsed:
        events.append(
            _progress_event(
                phase="waiting",
                worker_key=worker_key,
                backend_name=backend_name,
                elapsed_seconds=elapsed_seconds,
            ),
        )
        state.next_waiting_elapsed += _WAITING_PROGRESS_INTERVAL_SECONDS
    return events


def _next_progress_deadline_elapsed(state: _ProgressReporterState) -> float:
    if not state.cold_start_emitted:
        return _COLD_START_GRACE_SECONDS
    return state.next_waiting_elapsed


def _report_progress(progress_sink: ProgressSink, events: list[WorkerReadyProgress]) -> None:
    for event in events:
        progress_sink(event)


def _progress_terminal_event(
    *,
    state: _ProgressReporterState,
    phase: WorkerReadyPhase,
    worker_key: str,
    backend_name: str,
    error: str | None,
) -> WorkerReadyProgress | None:
    if phase == "ready" and not state.cold_start_emitted:
        return None
    return _progress_event(
        phase=phase,
        worker_key=worker_key,
        backend_name=backend_name,
        elapsed_seconds=max(0.0, time.monotonic() - state.started_at),
        error=error,
    )


def _progress_reporter_events(
    *,
    condition: threading.Condition,
    state: _ProgressReporterState,
    worker_key: str,
    backend_name: str,
) -> list[WorkerReadyProgress] | None:
    with condition:
        while True:
            if state.reporter_done:
                return None
            wait_timeout = state.started_at + _next_progress_deadline_elapsed(state) - time.monotonic()
            if wait_timeout > 0:
                condition.wait(timeout=wait_timeout)
                continue
            return _pending_progress_events(
                state=state,
                worker_key=worker_key,
                backend_name=backend_name,
            )


def _finalize_progress_events(
    *,
    condition: threading.Condition,
    state: _ProgressReporterState,
    phase: WorkerReadyPhase,
    worker_key: str,
    backend_name: str,
    error: str | None,
) -> tuple[list[WorkerReadyProgress], WorkerReadyProgress | None]:
    with condition:
        pending_events = _pending_progress_events(
            state=state,
            worker_key=worker_key,
            backend_name=backend_name,
        )
        terminal_event = _progress_terminal_event(
            state=state,
            phase=phase,
            worker_key=worker_key,
            backend_name=backend_name,
            error=error,
        )
        state.reporter_done = True
        condition.notify_all()
        return pending_events, terminal_event


def _progress_reporter_loop(
    *,
    condition: threading.Condition,
    state: _ProgressReporterState,
    progress_sink: ProgressSink,
    worker_key: str,
    backend_name: str,
) -> None:
    while True:
        pending_events = _progress_reporter_events(
            condition=condition,
            state=state,
            worker_key=worker_key,
            backend_name=backend_name,
        )
        if pending_events is None:
            return
        _report_progress(progress_sink, pending_events)


def _build_progress_reporter(
    *,
    worker_key: str,
    backend_name: str,
    progress_sink: ProgressSink,
) -> tuple[Callable[[float], None] | None, Callable[[WorkerReadyPhase, str | None], None]]:
    condition = threading.Condition()
    state = _ProgressReporterState(started_at=time.monotonic())

    thread = threading.Thread(
        target=_progress_reporter_loop,
        kwargs={
            "condition": condition,
            "state": state,
            "progress_sink": progress_sink,
            "worker_key": worker_key,
            "backend_name": backend_name,
        },
        name=f"kubernetes-worker-progress:{worker_key}",
        daemon=True,
    )
    thread.start()

    def on_poll_tick(_elapsed_seconds: float) -> None:
        with condition:
            condition.notify_all()

    def finalize(phase: WorkerReadyPhase, error: str | None) -> None:
        pending_events, terminal_event = _finalize_progress_events(
            condition=condition,
            state=state,
            phase=phase,
            worker_key=worker_key,
            backend_name=backend_name,
            error=error,
        )
        thread.join(timeout=_PROGRESS_REPORTER_JOIN_TIMEOUT_SECONDS)
        _report_progress(progress_sink, pending_events)
        if terminal_event is not None:
            progress_sink(terminal_event)

    return on_poll_tick, finalize


class KubernetesWorkerBackend:
    """Kubernetes-backed worker provider for dedicated worker pods."""

    backend_name = "kubernetes"

    def __init__(
        self,
        *,
        runtime_paths: RuntimePaths,
        config: KubernetesWorkerBackendConfig,
        auth_token: str | None,
        storage_root: Path,
        tool_validation_snapshot: dict[str, dict[str, object]],
        config_snapshot: dict[str, object],
        worker_grantable_credentials: frozenset[str],
    ) -> None:
        unsupported_services = sorted(
            {
                service
                for service in worker_grantable_credentials
                if not credential_service_policy(service, None).worker_grantable_supported
            },
        )
        if unsupported_services:
            msg = (
                "Dedicated workers do not support "
                f"{', '.join(unsupported_services)}. Keep these credentials in the primary runtime."
            )
            raise WorkerBackendError(msg)
        self.runtime_paths = runtime_paths
        self.config = config
        self.auth_token = auth_token
        self.storage_root = storage_root.expanduser().resolve()
        self.worker_grantable_credentials = worker_grantable_credentials
        self.idle_timeout_seconds = config.idle_timeout_seconds
        self._resources = resources.KubernetesResourceManager(
            runtime_paths=runtime_paths,
            config=config,
            auth_token=auth_token,
            storage_root=self.storage_root,
            tool_validation_snapshot=tool_validation_snapshot,
            config_snapshot=config_snapshot,
            worker_grantable_credentials=worker_grantable_credentials,
        )
        self._worker_locks: dict[str, threading.Lock] = {}
        self._worker_locks_lock = threading.Lock()
        self._progress_sinks: dict[str, list[ProgressSink]] = {}
        self._progress_snapshots: dict[str, WorkerReadyProgress | None] = {}
        self._progress_sinks_lock = threading.Lock()
        self._ready_workers: dict[str, _ReadyWorkerCacheEntry] = {}
        self._ready_workers_lock = threading.Lock()

    @classmethod
    def from_runtime(
        cls,
        runtime_paths: RuntimePaths,
        *,
        auth_token: str | None,
        storage_root: Path,
        tool_validation_snapshot: dict[str, dict[str, object]],
        config_snapshot: dict[str, object],
        worker_grantable_credentials: frozenset[str],
    ) -> KubernetesWorkerBackend:
        """Construct a backend instance from one explicit runtime context."""
        return cls(
            runtime_paths=runtime_paths,
            config=KubernetesWorkerBackendConfig.from_runtime(runtime_paths),
            auth_token=auth_token,
            storage_root=storage_root,
            tool_validation_snapshot=tool_validation_snapshot,
            config_snapshot=config_snapshot,
            worker_grantable_credentials=worker_grantable_credentials,
        )

    def shutdown(self) -> None:
        """Kubernetes workers are persistent; manager replacement leaves them running."""
        with self._ready_workers_lock:
            self._ready_workers.clear()

    def _register_progress_sink(self, worker_key: str, progress_sink: ProgressSink) -> None:
        with self._progress_sinks_lock:
            self._progress_sinks.setdefault(worker_key, []).append(progress_sink)
            snapshot = self._progress_snapshots.setdefault(worker_key, None)
            if snapshot is not None:
                progress_sink(snapshot)

    def _unregister_progress_sink(self, worker_key: str, progress_sink: ProgressSink) -> None:
        with self._progress_sinks_lock:
            sinks = self._progress_sinks.get(worker_key)
            if sinks is None:
                return
            for index, current_sink in enumerate(sinks):
                if current_sink is progress_sink:
                    sinks.pop(index)
                    break
            if not sinks:
                self._progress_sinks.pop(worker_key, None)
                self._progress_snapshots.pop(worker_key, None)

    def _emit_progress(self, progress: WorkerReadyProgress) -> None:
        with self._progress_sinks_lock:
            if progress.phase in {"ready", "failed"}:
                self._progress_snapshots.pop(progress.worker_key, None)
            else:
                self._progress_snapshots[progress.worker_key] = progress
            for current_sink in self._progress_sinks.get(progress.worker_key, ()):
                current_sink(progress)

    def ensure_worker(  # noqa: PLR0915
        self,
        spec: WorkerSpec,
        *,
        now: float | None = None,
        progress_sink: ProgressSink | None = None,
    ) -> WorkerHandle:
        """Resolve or start the worker backing the given worker key."""
        worker_key = spec.worker_key
        if progress_sink is not None:
            self._register_progress_sink(worker_key, progress_sink)
        try:
            with self._worker_lock(worker_key):
                timestamp = time.time() if now is None else now
                cached_handle = self._reuse_cached_ready_worker(spec, now=timestamp)
                if cached_handle is not None:
                    return cached_handle
                worker_id = self._worker_id(worker_key)
                state_subpath = self._state_subpath(worker_key)
                existing = self._resources.read_deployment(worker_id)
                current_handle = self._handle_from_deployment(existing, now=timestamp) if existing is not None else None
                should_restart = current_handle is None or current_handle.status in {"idle", "failed"}
                startup_count = (current_handle.startup_count if current_handle is not None else 0) + int(
                    should_restart,
                )
                created_at = current_handle.created_at if current_handle is not None else timestamp
                if should_restart:
                    last_started_at = timestamp
                else:
                    assert current_handle is not None
                    last_started_at = current_handle.last_started_at
                annotations = resources.metadata_annotations(
                    worker_key=worker_key,
                    state_subpath=state_subpath,
                    created_at=created_at,
                    last_used_at=timestamp,
                    last_started_at=last_started_at,
                    startup_count=startup_count,
                    failure_count=current_handle.failure_count if current_handle is not None else 0,
                    failure_reason=None,
                    status="starting",
                )
                should_report_progress = should_restart or existing is None or not self._deployment_ready(existing)
                poll_reporter: Callable[[float], None] | None = None
                finalize_progress: Callable[[WorkerReadyPhase, str | None], None] = _noop_finalize_progress
                destructive_failure_allowed = current_handle is None or current_handle.status != "ready"
                if should_report_progress:
                    poll_reporter, finalize_progress = _build_progress_reporter(
                        worker_key=worker_key,
                        backend_name=self.backend_name,
                        progress_sink=self._emit_progress,
                    )
                deployment_apply: resources.DeploymentApplyResult | None = None
                auth_secret_applied = False
                try:
                    self._resources.apply_auth_secret(worker_key=worker_key, worker_id=worker_id)
                    auth_secret_applied = True
                    deployment_apply = self._resources.apply_deployment(
                        worker_key=worker_key,
                        worker_id=worker_id,
                        state_subpath=state_subpath,
                        annotations=annotations,
                        replicas=1,
                        private_agent_names=spec.private_agent_names,
                    )
                    startup_triggered = should_restart or deployment_apply.recreated
                    destructive_failure_allowed = destructive_failure_allowed or startup_triggered
                    if startup_triggered and not should_report_progress:
                        poll_reporter, finalize_progress = _build_progress_reporter(
                            worker_key=worker_key,
                            backend_name=self.backend_name,
                            progress_sink=self._emit_progress,
                        )
                        should_report_progress = True
                    if deployment_apply.recreated and not should_restart:
                        startup_count = (current_handle.startup_count if current_handle is not None else 0) + 1
                        annotations = resources.metadata_annotations(
                            worker_key=worker_key,
                            state_subpath=state_subpath,
                            created_at=created_at,
                            last_used_at=timestamp,
                            last_started_at=timestamp,
                            startup_count=startup_count,
                            failure_count=current_handle.failure_count if current_handle is not None else 0,
                            failure_reason=None,
                            status="starting",
                        )
                        self._resources.patch_deployment(worker_id, annotations=annotations)
                    self._sync_shared_credentials(worker_key)
                    self._resources.apply_service(worker_id)
                    deployment = self._resources.wait_for_ready(
                        worker_id,
                        timeout_seconds=self.config.ready_timeout_seconds,
                        deployment_ready_fn=self._deployment_ready,
                        on_poll_tick=poll_reporter,
                    )
                    destructive_failure_allowed = False
                    final_annotations = dict(annotations)
                    final_annotations[resources.ANNOTATION_WORKER_STATUS] = "ready"
                    self._resources.patch_deployment(worker_id, annotations=final_annotations)
                except Exception as exc:
                    failure_reason = str(exc)
                    finalize_progress("failed", failure_reason)
                    self._record_startup_failure_or_cleanup_secret(
                        worker_key=worker_key,
                        worker_id=worker_id,
                        failure_reason=failure_reason,
                        timestamp=timestamp,
                        annotations=annotations,
                        destructive_failure_allowed=destructive_failure_allowed,
                        auth_secret_applied=auth_secret_applied,
                    )
                    if isinstance(exc, WorkerBackendError):
                        raise
                    raise WorkerBackendError(failure_reason) from exc

                finalize_progress("ready", None)
                final_deployment_annotations = {
                    **dict(deployment.metadata.annotations or {}),
                    **final_annotations,
                }
                handle = self._handle_from_deployment(
                    deployment,
                    now=timestamp,
                    annotations_override=final_deployment_annotations,
                )
                self._store_ready_worker(spec, handle, validated_at=timestamp)
                return handle
        finally:
            if progress_sink is not None:
                self._unregister_progress_sink(worker_key, progress_sink)

    def touch_worker(self, worker_key: str, *, now: float | None = None) -> WorkerHandle | None:
        """Refresh last-used metadata for one existing worker."""
        timestamp = time.time() if now is None else now
        with self._worker_lock(worker_key):
            cached_handle = self._touch_cached_worker(worker_key, now=timestamp)
            if cached_handle is not None:
                return cached_handle

            worker_id = self._worker_id(worker_key)
            deployment = self._resources.read_deployment(worker_id)
            if deployment is None:
                return None

            annotations = dict(deployment.metadata.annotations or {})
            resources.apply_lifecycle_annotations(
                annotations,
                touch_worker_lifecycle(
                    resources.lifecycle_from_annotations(annotations, now=timestamp),
                    now=timestamp,
                ),
            )
            self._resources.patch_deployment(worker_id, annotations=annotations)
            return self._handle_from_deployment(
                deployment,
                now=timestamp,
                annotations_override=annotations,
            )

    def list_workers(self, *, include_idle: bool = True, now: float | None = None) -> list[WorkerHandle]:
        """List workers known to this backend."""
        timestamp = time.time() if now is None else now
        handles = [
            self._handle_from_deployment(deployment, now=timestamp) for deployment in self._resources.list_deployments()
        ]
        return filter_and_sort_worker_handles(handles, include_idle)

    def cleanup_idle_workers(self, *, now: float | None = None) -> list[WorkerHandle]:
        """Scale idle workers to zero while retaining their state."""
        timestamp = time.time() if now is None else now
        return self._cleanup_idle_deployments(self._resources.list_deployments(), now=timestamp)

    def _cleanup_idle_deployments(
        self,
        deployments: list[resources.KubernetesDeployment],
        *,
        now: float,
    ) -> list[WorkerHandle]:
        """Scale idle workers from one already-loaded Deployment snapshot."""
        cleaned: list[WorkerHandle] = []
        for deployment in deployments:
            handle = self._handle_from_deployment(deployment, now=now)
            if handle.status != "idle" or int(deployment.spec.replicas or 0) == 0:
                continue
            annotations = dict(deployment.metadata.annotations or {})
            resources.apply_lifecycle_annotations(
                annotations,
                mark_worker_idle(resources.lifecycle_from_annotations(annotations, now=now)),
            )
            self._resources.patch_deployment(handle.worker_id, replicas=0, annotations=annotations)
            self._resources.delete_service(handle.worker_id)
            self._resources.delete_secret(handle.worker_id)
            self._invalidate_ready_worker(handle.worker_key)
            cleaned.append(
                self._handle_from_deployment(
                    deployment,
                    now=now,
                    annotations_override=annotations,
                    replicas_override=0,
                ),
            )
        return cleaned

    def _reconcile_drifted_deployments(
        self,
        deployments: list[resources.KubernetesDeployment],
        *,
        now: float,
        scaled_down_worker_ids: frozenset[str],
    ) -> list[WorkerHandle]:
        """Reconcile drifted workers from one already-loaded Deployment snapshot."""
        reconciled: list[WorkerHandle] = []
        for deployment in deployments:
            handle = self._handle_from_deployment(deployment, now=now)
            if not self._is_reconcile_candidate(
                deployment,
                handle=handle,
                scaled_down=handle.worker_id in scaled_down_worker_ids,
            ):
                continue
            worker_lock = self._worker_lock(handle.worker_key)
            if not worker_lock.acquire(blocking=False):
                continue
            try:
                refreshed = self._reconcile_worker_deployment(handle, now=now)
            finally:
                worker_lock.release()
            if refreshed is not None:
                reconciled.append(refreshed)
        return reconciled

    def maintain_workers(self, *, now: float | None = None) -> WorkerMaintenanceResult:
        """Clean and reconcile workers using one lightweight Deployment list."""
        timestamp = time.time() if now is None else now
        deployments = self._resources.list_deployments()
        cleaned = self._cleanup_idle_deployments(deployments, now=timestamp)
        reconciled = (
            self._reconcile_drifted_deployments(
                deployments,
                now=timestamp,
                scaled_down_worker_ids=frozenset(worker.worker_id for worker in cleaned),
            )
            if self.config.reconcile_pod_templates
            else []
        )
        return WorkerMaintenanceResult(cleaned=tuple(cleaned), reconciled=tuple(reconciled))

    def _is_reconcile_candidate(
        self,
        deployment: resources.KubernetesDeployment,
        *,
        handle: WorkerHandle,
        scaled_down: bool,
    ) -> bool:
        """Return whether one Deployment is a scaled-down worker with a drifted pod template.

        ``scaled_down`` marks workers this same maintenance pass just scaled to
        zero, whose snapshot still reports the pre-cleanup replica count.
        """
        if handle.status != "idle" or (not scaled_down and int(deployment.spec.replicas or 0) != 0):
            return False
        annotations = dict(deployment.metadata.annotations or {})
        private_agent_names = resources.parse_private_agent_names_annotation(annotations)
        if private_agent_names is None and resolved_worker_key_scope(handle.worker_key) == "user_agent":
            # Deployments created before visibility persistence cannot be rebuilt
            # deterministically; the ensure-time hash check recreates them on next use.
            logger.debug(
                "Deferring pod-template reconciliation for worker %r: no persisted private-agent visibility",
                handle.worker_key,
            )
            return False
        try:
            return self._resources.deployment_template_drifted(
                deployment,
                worker_key=handle.worker_key,
                worker_id=handle.worker_id,
                state_subpath=self._state_subpath(handle.worker_key),
                private_agent_names=private_agent_names,
            )
        except WorkerBackendError:
            logger.warning("Skipping pod-template reconciliation for worker %r", handle.worker_key, exc_info=True)
            return False

    def _reconcile_worker_deployment(self, handle: WorkerHandle, *, now: float) -> WorkerHandle | None:
        """Recreate one nominated worker after re-validating a fresh read under its lock.

        The list snapshot can be stale: a worker provisioned between the snapshot
        and the lock acquisition must not be scaled back down.
        """
        live = self._resources.read_deployment(handle.worker_id)
        if live is None:
            return None
        live_handle = self._handle_from_deployment(live, now=now)
        if not self._is_reconcile_candidate(live, handle=live_handle, scaled_down=False):
            return None
        annotations = dict(live.metadata.annotations or {})
        try:
            self._resources.apply_deployment(
                worker_key=live_handle.worker_key,
                worker_id=live_handle.worker_id,
                state_subpath=self._state_subpath(live_handle.worker_key),
                annotations=annotations,
                replicas=0,
                private_agent_names=resources.parse_private_agent_names_annotation(annotations),
            )
        except WorkerBackendError:
            logger.warning("Skipping pod-template reconciliation for worker %r", live_handle.worker_key, exc_info=True)
            return None
        # The recreate keeps the lifecycle annotations and zero replicas, so the
        # handle projected from the live read still describes the reconciled worker.
        return live_handle

    def record_failure(
        self,
        worker_key: str,
        failure_reason: str,
        *,
        now: float | None = None,
        annotations_override: dict[str, str] | None = None,
    ) -> WorkerHandle:
        """Persist a failed worker startup or execution state."""
        timestamp = time.time() if now is None else now
        with self._worker_lock(worker_key):
            return self._record_failure_locked(
                worker_key,
                failure_reason,
                now=timestamp,
                annotations_override=annotations_override,
            )

    def _record_failure_locked(
        self,
        worker_key: str,
        failure_reason: str,
        *,
        now: float,
        annotations_override: dict[str, str] | None = None,
    ) -> WorkerHandle:
        """Persist failure state while holding the worker provisioning lock."""
        self._invalidate_ready_worker(worker_key)
        worker_id = self._worker_id(worker_key)
        deployment = self._resources.read_deployment(worker_id)
        if deployment is None:
            msg = f"Unknown worker '{worker_key}' for Kubernetes failure recording."
            raise WorkerBackendError(msg)

        annotations = dict(deployment.metadata.annotations or {})
        if annotations_override is not None:
            annotations.update(annotations_override)
        resources.apply_lifecycle_annotations(
            annotations,
            mark_worker_failed(
                resources.lifecycle_from_annotations(annotations, now=now),
                now=now,
                failure_reason=failure_reason,
            ),
        )
        self._resources.patch_deployment(worker_id, replicas=0, annotations=annotations)
        self._resources.delete_service(worker_id)
        self._resources.delete_secret(worker_id)
        return self._handle_from_deployment(
            deployment,
            now=now,
            annotations_override=annotations,
            replicas_override=0,
        )

    def _sync_shared_credentials(self, worker_key: str) -> None:
        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=self.worker_grantable_credentials,
            credentials_manager=get_runtime_credentials_manager(self.runtime_paths),
        )

    def _reuse_cached_ready_worker(self, spec: WorkerSpec, *, now: float) -> WorkerHandle | None:
        entry = self._cached_ready_worker(spec.worker_key, spec=spec, now=now)
        if entry is None:
            return None
        try:
            handle = self._patch_cached_worker_usage(entry, now=now)
            if handle is None:
                return None
            self._sync_shared_credentials(spec.worker_key)
        except WorkerBackendError:
            self._invalidate_ready_worker(spec.worker_key)
            raise
        except Exception as exc:
            self._invalidate_ready_worker(spec.worker_key)
            raise WorkerBackendError(str(exc)) from exc
        return handle

    def _current_credentials_encryption_key_hash(self) -> str | None:
        encryption_key = credentials_encryption_key_value(
            self.runtime_paths.env_value(CREDENTIALS_ENCRYPTION_KEY_ENV),
        )
        return credentials_encryption_key_hash(encryption_key)

    def _store_ready_worker(
        self,
        spec: WorkerSpec,
        handle: WorkerHandle,
        *,
        validated_at: float,
    ) -> None:
        entry = _ReadyWorkerCacheEntry(
            spec=spec,
            handle=handle,
            validated_at=validated_at,
            credentials_encryption_key_hash=self._current_credentials_encryption_key_hash(),
        )
        with self._ready_workers_lock:
            self._ready_workers[spec.worker_key] = entry

    def _invalidate_ready_worker(self, worker_key: str) -> None:
        with self._ready_workers_lock:
            self._ready_workers.pop(worker_key, None)

    def _cached_ready_worker(
        self,
        worker_key: str,
        *,
        spec: WorkerSpec | None = None,
        now: float,
    ) -> _ReadyWorkerCacheEntry | None:
        with self._ready_workers_lock:
            entry = self._ready_workers.get(worker_key)
            if entry is None:
                return None
            cache_invalid = (
                (spec is not None and entry.spec != spec)
                or entry.credentials_encryption_key_hash != self._current_credentials_encryption_key_hash()
                or now - entry.validated_at >= _READY_WORKER_REVALIDATE_SECONDS
                or now - entry.handle.last_used_at >= self.idle_timeout_seconds
            )
            if cache_invalid:
                self._ready_workers.pop(worker_key, None)
                return None
            return entry

    def _touch_cached_worker(self, worker_key: str, *, now: float) -> WorkerHandle | None:
        entry = self._cached_ready_worker(worker_key, now=now)
        if entry is None:
            return None
        return self._patch_cached_worker_usage(entry, now=now)

    def _patch_cached_worker_usage(self, entry: _ReadyWorkerCacheEntry, *, now: float) -> WorkerHandle | None:
        handle = replace(entry.handle, last_used_at=now, status="ready", failure_reason=None)
        exists = self._resources.patch_deployment_annotations(
            handle.worker_id,
            annotations={
                resources.ANNOTATION_LAST_USED_AT: str(now),
                resources.ANNOTATION_WORKER_STATUS: "ready",
                resources.ANNOTATION_FAILURE_REASON: "",
            },
        )
        if not exists:
            self._invalidate_ready_worker(entry.spec.worker_key)
            return None
        self._store_ready_worker(
            entry.spec,
            handle,
            validated_at=entry.validated_at,
        )
        return handle

    def _record_startup_failure_or_cleanup_secret(
        self,
        *,
        worker_key: str,
        worker_id: str,
        failure_reason: str,
        timestamp: float,
        annotations: dict[str, str],
        destructive_failure_allowed: bool,
        auth_secret_applied: bool,
    ) -> None:
        deployment_after_failure = self._resources.read_deployment(worker_id)
        if deployment_after_failure is not None:
            if destructive_failure_allowed:
                self._record_failure_locked(
                    worker_key,
                    failure_reason,
                    now=timestamp,
                    annotations_override=annotations,
                )
        elif auth_secret_applied:
            self._resources.delete_secret(worker_id)

    def _worker_lock(self, worker_key: str) -> threading.Lock:
        with self._worker_locks_lock:
            worker_lock = self._worker_locks.get(worker_key)
            if worker_lock is None:
                worker_lock = threading.Lock()
                self._worker_locks[worker_key] = worker_lock
        return worker_lock

    def _worker_id(self, worker_key: str) -> str:
        return worker_id_for_key(worker_key, prefix=self.config.name_prefix)

    def _state_subpath(self, worker_key: str) -> str:
        prefix = self.config.storage_subpath_prefix.strip().strip("/")
        worker_dir = worker_dir_name(worker_key)
        return f"{prefix}/{worker_dir}" if prefix else worker_dir

    def _deployment_ready(self, deployment: resources.KubernetesDeployment) -> bool:
        desired = int(deployment.spec.replicas or 0)
        if desired == 0:
            return True
        ready = int(deployment.status.ready_replicas or 0)
        observed_generation = deployment.status.observed_generation
        generation = deployment.metadata.generation
        generation_ready = observed_generation is None or generation is None or observed_generation >= generation
        return generation_ready and ready >= desired

    def _handle_from_deployment(
        self,
        deployment: resources.KubernetesDeployment,
        *,
        now: float,
        annotations_override: dict[str, str] | None = None,
        replicas_override: int | None = None,
    ) -> WorkerHandle:
        metadata = deployment.metadata
        annotations = dict(metadata.annotations or {}) if annotations_override is None else dict(annotations_override)
        worker_key = annotations.get(resources.ANNOTATION_WORKER_KEY)
        if not worker_key:
            msg = f"Deployment '{metadata.name}' is missing worker metadata."
            raise WorkerBackendError(msg)

        worker_id = str(metadata.name)
        last_used_at = resources.parse_annotation_float(annotations, resources.ANNOTATION_LAST_USED_AT, now)
        created_at = resources.parse_annotation_float(annotations, resources.ANNOTATION_CREATED_AT, last_used_at)
        last_started_at = annotations.get(resources.ANNOTATION_LAST_STARTED_AT)
        status = self._effective_status(
            deployment,
            now=now,
            annotations_override=annotations,
            replicas_override=replicas_override,
        )
        endpoint_root = resources.service_host(worker_id, self.config.namespace, self.config.worker_port)
        return WorkerHandle(
            worker_id=worker_id,
            worker_key=worker_key,
            endpoint=f"{endpoint_root}/api/sandbox-runner/execute",
            auth_token=resources.worker_auth_token(self.auth_token, worker_key),
            status=status,
            backend_name=self.backend_name,
            last_used_at=last_used_at,
            created_at=created_at,
            last_started_at=float(last_started_at) if last_started_at is not None else None,
            expires_at=None,
            startup_count=resources.parse_annotation_int(annotations, resources.ANNOTATION_STARTUP_COUNT),
            failure_count=resources.parse_annotation_int(annotations, resources.ANNOTATION_FAILURE_COUNT),
            failure_reason=annotations.get(resources.ANNOTATION_FAILURE_REASON) or None,
            debug_metadata={
                "namespace": self.config.namespace,
                "deployment_name": worker_id,
                "service_name": worker_id,
                "state_subpath": annotations.get(resources.ANNOTATION_STATE_SUBPATH, ""),
                "api_root": f"{endpoint_root}/api/sandbox-runner",
            },
        )

    def _effective_status(
        self,
        deployment: resources.KubernetesDeployment,
        *,
        now: float,
        annotations_override: dict[str, str] | None = None,
        replicas_override: int | None = None,
    ) -> WorkerStatus:
        annotations = (
            dict(deployment.metadata.annotations or {}) if annotations_override is None else annotations_override
        )
        stored_status = annotations.get(resources.ANNOTATION_WORKER_STATUS, "starting")
        if stored_status == "failed":
            return "failed"
        replicas = int(deployment.spec.replicas or 0) if replicas_override is None else replicas_override
        if replicas == 0:
            return "idle"
        if not self._deployment_ready(deployment):
            return "starting"
        last_used_at = resources.parse_annotation_float(annotations, resources.ANNOTATION_LAST_USED_AT, now)
        return effective_idle_status("ready", last_used_at, self.idle_timeout_seconds, now)
