"""Docker-backed worker backend for dedicated local worker containers."""

from __future__ import annotations

import hashlib
import importlib
import json
import os
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Protocol, cast

import httpx
import yaml

from mindroom.config.yaml_includes import load_yaml_config_source_with_digests, source_files_fingerprint
from mindroom.constants import (
    DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    RuntimePaths,
    resolve_primary_runtime_paths,
    runtime_paths_with_config_path,
    runtime_paths_with_storage_root,
    serialize_runtime_paths,
)
from mindroom.credentials import CredentialsManager, get_runtime_credentials_manager, sync_shared_credentials_to_worker
from mindroom.redaction import redact_sensitive_text
from mindroom.runtime_env_policy import SANDBOX_RUNTIME_ENV_BY_KEY, SHARED_CREDENTIALS_PATH_ENV
from mindroom.tool_system.dependencies import ensure_optional_deps
from mindroom.tool_system.worker_routing import resolved_worker_key_scope, worker_dir_name, worker_key_agent_name
from mindroom.workers.backend import WorkerBackendError
from mindroom.workers.backends._dedicated_worker_common import (
    build_dedicated_worker_runtime_paths,
    plan_scoped_visible_state_roots,
    validate_dedicated_worker_extra_env,
    validate_unique_worker_visible_paths,
)
from mindroom.workers.backends._lifecycle import (
    initial_worker_lifecycle_state,
    mark_worker_failed,
    mark_worker_idle,
    mark_worker_ready,
    prepare_worker_ensure_lifecycle,
    read_lifecycle_state,
    touch_worker_lifecycle,
    write_lifecycle_state,
)
from mindroom.workers.backends._metadata_store import (
    list_worker_state_paths,
    load_worker_metadata,
    save_worker_metadata,
)
from mindroom.workers.backends.docker_config import (
    DEFAULT_WORKER_PORT,
    DOCKER_RESERVED_EXTRA_ENV_NAMES,
    DockerWorkerBackendConfig,
    docker_backend_config_signature,
    docker_workers_root,
    normalize_docker_name_prefix,
    resolve_docker_storage_path,
)
from mindroom.workers.backends.docker_projection import PROJECTED_CONFIGS_DIRNAME, DockerProjectionManager
from mindroom.workers.backends.local import LocalWorkerStatePaths, local_worker_state_paths_for_root
from mindroom.workers.compatibility import WORKER_PROTOCOL_VERSION
from mindroom.workers.models import (
    ProgressSink,
    WorkerHandle,
    WorkerReadyPhase,
    WorkerReadyProgress,
    WorkerSpec,
    WorkerStatus,
)

if TYPE_CHECKING:
    from collections.abc import Callable

    class _DockerContainer(Protocol):
        attrs: dict[str, object]
        status: str
        id: str

        def reload(self) -> None: ...

        def start(self) -> None: ...

        def stop(self, timeout: int = 10) -> None: ...

        def remove(self, force: bool = True) -> None: ...

        def logs(self, *, tail: int = ...) -> bytes: ...

    class _DockerContainersApi(Protocol):
        def get(self, name: str) -> _DockerContainer: ...

        def run(self, image: str, **kwargs: object) -> _DockerContainer: ...

    class _DockerImage(Protocol):
        id: str

    class _DockerImagesApi(Protocol):
        def get(self, name: str) -> _DockerImage: ...

        def pull(self, name: str) -> _DockerImage: ...

    class _DockerClient(Protocol):
        containers: _DockerContainersApi
        images: _DockerImagesApi

    class _DockerErrors(Protocol):
        DockerException: type[Exception]
        NotFound: type[Exception]


_READY_POLL_INTERVAL_SECONDS = 1.0
_CONTAINER_LOG_EXCERPT_MAX_CHARS = 4096


class _WorkerImageIncompatibleError(WorkerBackendError):
    """A ready worker answered the health check with another protocol."""


_TOKEN_ENV_NAME = SANDBOX_RUNTIME_ENV_BY_KEY["proxy_token"]
_RUNNER_PORT_ENV_NAME = SANDBOX_RUNTIME_ENV_BY_KEY["runner_port"]
_STARTUP_RUNTIME_PATHS_ENV = "MINDROOM_RUNTIME_PATHS_JSON"
_DEDICATED_WORKER_KEY_ENV = SANDBOX_RUNTIME_ENV_BY_KEY["dedicated_worker_key"]
_DEDICATED_WORKER_ROOT_ENV = SANDBOX_RUNTIME_ENV_BY_KEY["dedicated_worker_root"]
_SHARED_STORAGE_ROOT_ENV = SANDBOX_RUNTIME_ENV_BY_KEY["shared_storage_root"]

_LABEL_COMPONENT = "mindroom.ai/component"
_LABEL_COMPONENT_VALUE = "worker"
_LABEL_MANAGED_BY = "app.mindroom.ai/managed-by"
_LABEL_MANAGED_BY_VALUE = "mindroom"
_LABEL_NAME = "app.mindroom.ai/name"
_LABEL_NAME_VALUE = "mindroom-docker-worker"
_LABEL_WORKER_ID = "mindroom.ai/worker-id"
_LABEL_LAUNCH_CONFIG_HASH = "mindroom.ai/launch-config-hash"
_LABEL_RUNTIME_NAMESPACE = "mindroom.ai/runtime-namespace"

_DOCKER_DEPENDENCIES = ["docker"]
_DOCKER_EXTRA = "docker"

__all__ = [
    "DockerWorkerBackend",
    "docker_backend_config_signature",
    "ensure_docker_dependencies",
]


def _runtime_namespace_for_workers_root(workers_root: Path) -> str:
    resolved_workers_root = workers_root.expanduser().resolve()
    return hashlib.sha256(str(resolved_workers_root).encode("utf-8")).hexdigest()[:12]


def _container_name_for_worker(worker_key: str, *, prefix: str, runtime_namespace: str) -> str:
    digest = hashlib.sha256(f"{runtime_namespace}:{worker_key}".encode()).hexdigest()[:24]
    normalized_prefix = normalize_docker_name_prefix(prefix)
    max_prefix_length = 63 - len(digest) - 1
    safe_prefix = normalized_prefix[:max_prefix_length].rstrip("-")
    if not safe_prefix:
        safe_prefix = normalize_docker_name_prefix("mindroom-worker")[:max_prefix_length].rstrip("-") or "worker"
    return f"{safe_prefix}-{digest}"


def _host_config_contents_hash(host_config_path: Path | None) -> str:
    if host_config_path is None:
        return ""
    try:
        _, source_digests = load_yaml_config_source_with_digests(host_config_path)
    except OSError as exc:
        msg = f"Failed to read Docker worker config file '{host_config_path}': {exc}"
        raise WorkerBackendError(msg) from exc
    except (yaml.YAMLError, UnicodeError):
        # An unparseable config still hashes by its top-level bytes so worker
        # staleness checks keep working while the user fixes the file.
        try:
            return hashlib.sha256(host_config_path.read_bytes()).hexdigest()
        except OSError as exc:
            msg = f"Failed to read Docker worker config file '{host_config_path}': {exc}"
            raise WorkerBackendError(msg) from exc
    return source_files_fingerprint(host_config_path, source_digests)


def _docker_image_identity_state(
    image: str,
    *,
    client: _DockerClient,
    docker_errors: _DockerErrors,
) -> tuple[str, bool]:
    try:
        docker_image = client.images.get(image)
    except docker_errors.NotFound:
        return image, False
    except docker_errors.DockerException:
        return image, False

    image_id = docker_image.id
    if isinstance(image_id, str) and image_id.strip():
        return image_id, True
    return image, False


def _resolved_docker_image_identity(
    image: str,
    *,
    client: _DockerClient,
    docker_errors: _DockerErrors,
) -> str:
    resolved_identity, _ = _docker_image_identity_state(
        image,
        client=client,
        docker_errors=docker_errors,
    )
    return resolved_identity


def _worker_health_compatibility_error(response: httpx.Response, *, image: str) -> str | None:
    """Return an actionable error when a ready worker speaks another protocol."""
    try:
        payload = response.json()
    except ValueError:
        payload = None

    if not isinstance(payload, dict):
        reported_protocol: object = "missing"
        reported_version: object = "unknown"
    else:
        reported_protocol = payload.get("worker_protocol", "missing")
        reported_version = payload.get("mindroom_version", "unknown")

    if (
        isinstance(reported_protocol, int)
        and not isinstance(reported_protocol, bool)
        and reported_protocol == WORKER_PROTOCOL_VERSION
    ):
        return None
    return (
        f"Docker worker image '{image}' is incompatible with this MindRoom runtime: "
        f"expected worker protocol {WORKER_PROTOCOL_VERSION}, got {reported_protocol} "
        f"(worker MindRoom version {reported_version}). "
        "Use a worker image built for this MindRoom release."
    )


def ensure_docker_dependencies(runtime_paths: RuntimePaths | None = None) -> None:
    """Install the optional Docker SDK runtime when needed."""
    effective_runtime_paths = runtime_paths or resolve_primary_runtime_paths(process_env=dict(os.environ))
    try:
        ensure_optional_deps(
            _DOCKER_DEPENDENCIES,
            _DOCKER_EXTRA,
            effective_runtime_paths,
        )
    except ImportError as exc:
        raise WorkerBackendError(str(exc)) from exc


def _load_docker_client_and_errors(
    *,
    runtime_paths: RuntimePaths | None = None,
) -> tuple[_DockerClient, _DockerErrors]:
    ensure_docker_dependencies(runtime_paths)
    try:
        docker_module = importlib.import_module("docker")
        docker_errors = cast("_DockerErrors", importlib.import_module("docker.errors"))
    except ModuleNotFoundError as exc:
        msg = "The Docker worker backend could not import the Docker SDK after ensuring the optional 'docker' extra."
        raise WorkerBackendError(msg) from exc

    docker_from_env = cast("Callable[[], _DockerClient]", docker_module.from_env)
    try:
        client = docker_from_env()
    except docker_errors.DockerException as exc:
        msg = f"Failed to initialize Docker client: {exc}"
        raise WorkerBackendError(msg) from exc
    return client, docker_errors


@dataclass
class _DockerWorkerMetadata:
    worker_id: str
    worker_key: str
    endpoint: str
    backend_name: str
    container_name: str
    created_at: float
    last_used_at: float
    status: WorkerStatus
    host_port: int | None = None
    container_id: str | None = None
    image: str | None = None
    publish_host: str | None = None
    worker_port: int = DEFAULT_WORKER_PORT
    last_started_at: float | None = None
    startup_count: int = 0
    failure_count: int = 0
    failure_reason: str | None = None
    launch_config_hash: str | None = None


@dataclass(frozen=True, slots=True)
class _DockerLaunchConfig:
    image_reference: str
    launch_config_hash: str


class DockerWorkerBackend:
    """Docker-backed worker provider for dedicated local sandbox-runner containers."""

    backend_name = "docker"

    def __init__(
        self,
        *,
        config: DockerWorkerBackendConfig,
        auth_token: str | None,
        storage_path: Path | None = None,
        runtime_paths: RuntimePaths | None = None,
        worker_grantable_credentials: frozenset[str] = DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    ) -> None:
        if auth_token is None:
            msg = "A worker auth token is required for Docker workers."
            raise WorkerBackendError(msg)
        validate_dedicated_worker_extra_env(
            config.extra_env,
            backend_name="Docker",
            extra_reserved_names=DOCKER_RESERVED_EXTRA_ENV_NAMES,
        )

        self.config = config
        self.auth_token = auth_token
        self.idle_timeout_seconds = config.idle_timeout_seconds
        self._storage_path = resolve_docker_storage_path(storage_path, runtime_paths=runtime_paths)
        self._workers_root = docker_workers_root(self._storage_path)
        base_runtime_paths = (
            resolve_primary_runtime_paths(
                config_path=config.host_config_path,
                storage_path=self._storage_path,
                process_env=dict(os.environ),
            )
            if runtime_paths is None
            else runtime_paths
        )
        if config.host_config_path is not None:
            base_runtime_paths = runtime_paths_with_config_path(base_runtime_paths, config.host_config_path)
        self._runtime_paths = runtime_paths_with_storage_root(base_runtime_paths, self._storage_path)
        self._client, self._docker_errors = _load_docker_client_and_errors(runtime_paths=self._runtime_paths)
        self._worker_locks: dict[str, threading.Lock] = {}
        self._worker_locks_lock = threading.Lock()
        self._metadata_lock = threading.Lock()
        self._projection_manager = DockerProjectionManager(
            config=config,
            projected_configs_root=self._workers_root / PROJECTED_CONFIGS_DIRNAME,
            runtime_paths=self._runtime_paths,
        )
        if runtime_paths is None:
            self._credentials_manager = CredentialsManager(base_path=self._storage_path / "credentials")
        else:
            self._credentials_manager = get_runtime_credentials_manager(self._runtime_paths)
        self.worker_grantable_credentials = worker_grantable_credentials
        self._runtime_namespace = _runtime_namespace_for_workers_root(self._workers_root)
        self._workers_root.mkdir(parents=True, exist_ok=True)

    @classmethod
    def from_runtime(
        cls,
        runtime_paths: RuntimePaths,
        *,
        auth_token: str | None,
        storage_path: Path | None = None,
        worker_grantable_credentials: frozenset[str] = DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    ) -> DockerWorkerBackend:
        """Construct a backend instance from one explicit runtime context."""
        return cls(
            config=DockerWorkerBackendConfig.from_runtime(runtime_paths),
            auth_token=auth_token,
            storage_path=storage_path,
            runtime_paths=runtime_paths,
            worker_grantable_credentials=worker_grantable_credentials,
        )

    def shutdown(self) -> None:
        """Remove backend-owned containers before discarding this Docker manager."""
        failures: list[str] = []
        for paths in self._metadata_paths():
            metadata = self._load_metadata(paths)
            if metadata is None:
                continue
            with self._worker_lock(metadata.worker_key):
                metadata = self._load_metadata(paths)
                if metadata is None:
                    continue
                try:
                    self._remove_container(self._read_container(metadata.container_name))
                except WorkerBackendError as exc:
                    failures.append(str(exc))
                    continue
                write_lifecycle_state(
                    metadata,
                    mark_worker_idle(read_lifecycle_state(metadata)),
                )
                metadata.endpoint = self._endpoint_for_host_port(None)
                metadata.host_port = None
                metadata.container_id = None
                metadata.launch_config_hash = None
                self._save_metadata(paths, metadata)
        if failures:
            failure_text = "; ".join(failures)
            msg = f"Failed to shut down Docker workers: {failure_text}"
            raise WorkerBackendError(msg)

    def ensure_worker(
        self,
        spec: WorkerSpec,
        *,
        now: float | None = None,
        progress_sink: ProgressSink | None = None,
    ) -> WorkerHandle:
        """Resolve or start the dedicated worker container for the given worker key."""
        timestamp = time.time() if now is None else now
        start_time = time.monotonic()

        def emit_progress(phase: WorkerReadyPhase, *, error: str | None = None) -> None:
            if progress_sink is None:
                return
            progress_sink(
                WorkerReadyProgress(
                    phase=phase,
                    worker_key=spec.worker_key,
                    backend_name=self.backend_name,
                    elapsed_seconds=max(0.0, time.monotonic() - start_time),
                    error=error,
                ),
            )

        with self._worker_lock(spec.worker_key):
            launch_config = self._resolve_launch_config()
            paths = self._state_paths(spec.worker_key)
            metadata = self._load_metadata(paths) or self._default_metadata(
                spec.worker_key,
                timestamp,
                launch_config_hash=launch_config.launch_config_hash,
            )
            identity_changed = self._sync_metadata_identity(metadata)

            should_restart = identity_changed or self._should_restart(
                metadata,
                paths,
                private_agent_names=spec.private_agent_names,
            )
            lifecycle_state = prepare_worker_ensure_lifecycle(
                read_lifecycle_state(metadata),
                now=timestamp,
                should_restart=should_restart,
            )
            write_lifecycle_state(metadata, lifecycle_state)
            self._save_metadata(paths, metadata)
            cold_start_emitted = lifecycle_state.status == "starting"
            if cold_start_emitted:
                emit_progress("cold_start")

            sync_shared_credentials_to_worker(
                spec.worker_key,
                allowed_services=self.worker_grantable_credentials,
                credentials_manager=self._credentials_manager,
            )

            try:
                container = self._ensure_container(
                    metadata,
                    paths,
                    private_agent_names=spec.private_agent_names,
                    launch_config=launch_config,
                )
                try:
                    endpoint = self._wait_for_ready(container)
                except _WorkerImageIncompatibleError as stale_error:
                    write_lifecycle_state(
                        metadata,
                        prepare_worker_ensure_lifecycle(
                            read_lifecycle_state(metadata),
                            now=timestamp,
                            should_restart=True,
                        ),
                    )
                    self._save_metadata(paths, metadata)
                    if not cold_start_emitted:
                        emit_progress("cold_start")
                    container, launch_config = self._relaunch_after_stale_image(
                        metadata,
                        paths,
                        private_agent_names=spec.private_agent_names,
                        stale_error=stale_error,
                    )
                    endpoint = self._wait_for_ready(container)
            except Exception as exc:
                failure_reason = str(exc)
                self._record_failure_locked(paths, metadata, failure_reason, now=timestamp, stop_container=True)
                emit_progress("failed", error=failure_reason)
                if isinstance(exc, WorkerBackendError):
                    raise
                raise WorkerBackendError(failure_reason) from exc

            write_lifecycle_state(
                metadata,
                mark_worker_ready(read_lifecycle_state(metadata), now=timestamp),
            )
            metadata.endpoint = endpoint
            metadata.host_port = self._container_host_port(container)
            metadata.container_id = self._container_id(container)
            metadata.image = self.config.image
            metadata.publish_host = self.config.publish_host
            metadata.worker_port = self.config.worker_port
            metadata.launch_config_hash = launch_config.launch_config_hash
            self._save_metadata(paths, metadata)
            handle = self._to_handle(metadata, container, now=timestamp, paths=paths)
            emit_progress("ready")
            return handle

    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):
            paths = self._state_paths(worker_key)
            metadata = self._load_metadata(paths)
            if metadata is None:
                return None
            write_lifecycle_state(
                metadata,
                touch_worker_lifecycle(read_lifecycle_state(metadata), now=timestamp),
            )
            container = self._read_container(metadata.container_name)
            metadata = self._reconcile_missing_container_metadata(paths, metadata, container)
            self._save_metadata(paths, metadata)
            return self._to_handle(metadata, container, now=timestamp, paths=paths)

    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: list[WorkerHandle] = []
        for paths in self._metadata_paths():
            metadata = self._load_metadata(paths)
            if metadata is None:
                continue
            with self._worker_lock(metadata.worker_key):
                metadata = self._load_metadata(paths)
                if metadata is None:
                    continue
                container = self._read_container(metadata.container_name)
                metadata = self._reconcile_missing_container_metadata(paths, metadata, container)
                handle = self._to_handle(
                    metadata,
                    container,
                    now=timestamp,
                    paths=paths,
                )
            if include_idle or handle.status != "idle":
                handles.append(handle)
        return sorted(handles, key=lambda handle: handle.last_used_at, reverse=True)

    def cleanup_idle_workers(self, *, now: float | None = None) -> list[WorkerHandle]:
        """Stop idle containers while retaining worker-owned state."""
        timestamp = time.time() if now is None else now
        cleaned: list[WorkerHandle] = []
        for paths in self._metadata_paths():
            metadata = self._load_metadata(paths)
            if metadata is None:
                continue
            with self._worker_lock(metadata.worker_key):
                metadata = self._load_metadata(paths)
                if metadata is None:
                    continue
                container = self._read_container(metadata.container_name)
                metadata = self._reconcile_missing_container_metadata(paths, metadata, container)
                handle = self._to_handle(metadata, container, now=timestamp, paths=paths)
                idle_timed_out = timestamp - metadata.last_used_at >= self.idle_timeout_seconds
                if handle.status == "idle" and self._container_is_running(container):
                    self._stop_container(container)
                    write_lifecycle_state(
                        metadata,
                        mark_worker_idle(read_lifecycle_state(metadata)),
                    )
                    self._save_metadata(paths, metadata)
                    cleaned.append(self._to_handle(metadata, container, now=timestamp, paths=paths))
                elif handle.status == "failed" and container is not None and idle_timed_out:
                    # A worker that failed and was never revived keeps its exited
                    # container so a quick retry can restart it. Once it is past the
                    # idle timeout it is abandoned, so reap the container to stop
                    # stale failures from accumulating; the failed metadata is kept
                    # and a later ensure recreates the container.
                    self._stop_container(container)
                    self._remove_container(container)
                    self._reconcile_missing_container_metadata(paths, metadata, None)
        return sorted(cleaned, key=lambda handle: handle.last_used_at, reverse=True)

    def record_failure(self, worker_key: str, failure_reason: str, *, now: float | 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):
            paths = self._state_paths(worker_key)
            metadata = self._load_metadata(paths) or self._default_metadata(worker_key, timestamp)
            return self._record_failure_locked(paths, metadata, failure_reason, now=timestamp, stop_container=True)

    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 _state_paths(self, worker_key: str) -> LocalWorkerStatePaths:
        return local_worker_state_paths_for_root(self._workers_root / worker_dir_name(worker_key))

    def _default_metadata(
        self,
        worker_key: str,
        now: float,
        *,
        launch_config_hash: str | None = None,
    ) -> _DockerWorkerMetadata:
        worker_id = self._container_name_for_worker(worker_key)
        lifecycle = initial_worker_lifecycle_state(now=now)
        resolved_launch_config_hash = (
            self._compute_launch_config_hash() if launch_config_hash is None else launch_config_hash
        )
        return _DockerWorkerMetadata(
            worker_id=worker_id,
            worker_key=worker_key,
            endpoint=self._endpoint_for_host_port(None),
            backend_name=self.backend_name,
            container_name=worker_id,
            created_at=lifecycle.created_at,
            last_used_at=lifecycle.last_used_at,
            status=lifecycle.status,
            image=self.config.image,
            publish_host=self.config.publish_host,
            worker_port=self.config.worker_port,
            launch_config_hash=resolved_launch_config_hash,
        )

    def _container_name_for_worker(self, worker_key: str) -> str:
        return _container_name_for_worker(
            worker_key,
            prefix=self.config.name_prefix,
            runtime_namespace=self._runtime_namespace,
        )

    def _sync_metadata_identity(self, metadata: _DockerWorkerMetadata) -> bool:
        expected_container_name = self._container_name_for_worker(metadata.worker_key)
        if metadata.container_name == expected_container_name and metadata.worker_id == expected_container_name:
            return False

        if metadata.container_name != expected_container_name:
            self._remove_container(self._read_container(metadata.container_name))
        metadata.worker_id = expected_container_name
        metadata.container_name = expected_container_name
        metadata.endpoint = self._endpoint_for_host_port(None)
        metadata.host_port = None
        metadata.container_id = None
        metadata.launch_config_hash = None
        return True

    def _metadata_paths(self) -> list[LocalWorkerStatePaths]:
        return list_worker_state_paths(
            self._workers_root,
            state_paths_from_root=local_worker_state_paths_for_root,
        )

    def _load_metadata(self, paths: LocalWorkerStatePaths) -> _DockerWorkerMetadata | None:
        return load_worker_metadata(paths, metadata_type=_DockerWorkerMetadata)

    def _save_metadata(self, paths: LocalWorkerStatePaths, metadata: _DockerWorkerMetadata) -> None:
        save_worker_metadata(
            paths,
            metadata,
            ensure_root=True,
            lock=self._metadata_lock,
        )

    def _reconcile_missing_container_metadata(
        self,
        paths: LocalWorkerStatePaths,
        metadata: _DockerWorkerMetadata,
        container: _DockerContainer | None,
    ) -> _DockerWorkerMetadata:
        if container is not None:
            return metadata
        if metadata.host_port is None and metadata.container_id is None:
            return metadata
        metadata.endpoint = self._endpoint_for_host_port(None)
        metadata.host_port = None
        metadata.container_id = None
        self._save_metadata(paths, metadata)
        return metadata

    def _read_container(self, container_name: str) -> _DockerContainer | None:
        try:
            return self._client.containers.get(container_name)
        except self._docker_errors.NotFound:
            return None
        except self._docker_errors.DockerException as exc:
            msg = f"Failed to inspect Docker worker '{container_name}': {exc}"
            raise WorkerBackendError(msg) from exc

    def _should_restart(
        self,
        metadata: _DockerWorkerMetadata,
        paths: LocalWorkerStatePaths,
        *,
        private_agent_names: frozenset[str] | None,
    ) -> bool:
        container = self._read_container(metadata.container_name)
        if metadata.status == "failed":
            return True
        if container is None:
            return True
        if not self._container_matches_config(
            metadata,
            container,
            paths,
            private_agent_names=private_agent_names,
        ):
            return True
        return not self._container_is_running(container)

    def _container_matches_config(
        self,
        metadata: _DockerWorkerMetadata,
        container: _DockerContainer | None,
        paths: LocalWorkerStatePaths,
        *,
        private_agent_names: frozenset[str] | None,
    ) -> bool:
        compatible_launch_config_hashes = self._compatible_launch_config_hashes(container)
        if metadata.launch_config_hash not in compatible_launch_config_hashes:
            return False
        if self._container_launch_config_hash(container) not in compatible_launch_config_hashes:
            return False

        config_mount_specs, projection = self._projection_manager.config_mount_specs(
            paths,
            worker_key=metadata.worker_key,
            materialize_projection=False,
        )
        if projection is not None and not projection.ready:
            return False

        if not self._container_env_matches(
            container,
            expected_env=self._container_env(metadata.worker_key),
        ):
            return False

        mount_checks = [
            (paths.root, self.config.storage_mount_path, False),
        ]
        mount_checks.extend(
            self._scoped_storage_mount_specs(
                metadata.worker_key,
                private_agent_names=private_agent_names,
            ),
        )
        mount_checks.extend(config_mount_specs)
        return self._container_mount_layout_matches(container, expected_mounts=mount_checks)

    def _ensure_container(
        self,
        metadata: _DockerWorkerMetadata,
        paths: LocalWorkerStatePaths,
        *,
        private_agent_names: frozenset[str] | None,
        launch_config: _DockerLaunchConfig,
    ) -> _DockerContainer:
        paths.root.mkdir(parents=True, exist_ok=True)
        container = self._read_container(metadata.container_name)
        if container is not None and not self._container_matches_config(
            metadata,
            container,
            paths,
            private_agent_names=private_agent_names,
        ):
            self._remove_container(container)
            container = None

        if container is None:
            container = self._client.containers.run(
                launch_config.image_reference,
                command=["/app/run-sandbox-runner.sh"],
                name=metadata.container_name,
                detach=True,
                environment=self._container_env(metadata.worker_key),
                volumes=self._container_volumes(
                    paths,
                    worker_key=metadata.worker_key,
                    private_agent_names=private_agent_names,
                ),
                ports={f"{self.config.worker_port}/tcp": (self.config.publish_host, None)},
                labels=self._container_labels(
                    metadata,
                    launch_config_hash=launch_config.launch_config_hash,
                ),
                user=self.config.user,
            )
        elif not self._container_is_running(container):
            try:
                container.start()
            except self._docker_errors.DockerException as exc:
                msg = f"Failed to start Docker worker '{metadata.container_name}': {exc}"
                raise WorkerBackendError(msg) from exc

        self._reload_container(container)
        if self._container_host_port(container) is None:
            msg = f"Docker worker '{metadata.container_name}' is missing a published port."
            raise WorkerBackendError(msg)
        return container

    def _relaunch_after_stale_image(
        self,
        metadata: _DockerWorkerMetadata,
        paths: LocalWorkerStatePaths,
        *,
        private_agent_names: frozenset[str] | None,
        stale_error: WorkerBackendError,
    ) -> tuple[_DockerContainer, _DockerLaunchConfig]:
        """Pull the configured image once and relaunch after a protocol mismatch.

        A stale local image otherwise fails every worker start until someone
        pulls by hand, since tag references like ``:latest`` keep resolving to
        the local image.
        """
        try:
            pulled_image = self._client.images.pull(self.config.image)
        except self._docker_errors.DockerException as exc:
            msg = f"{stale_error} Pulling '{self.config.image}' failed: {exc}"
            raise WorkerBackendError(msg) from exc
        launch_config = self._launch_config_from_image_identity(
            pulled_image.id,
            image_resolved=True,
        )
        metadata.launch_config_hash = launch_config.launch_config_hash
        self._save_metadata(paths, metadata)
        container = self._read_container(metadata.container_name)
        self._stop_container(container)
        self._remove_container(container)
        return (
            self._ensure_container(
                metadata,
                paths,
                private_agent_names=private_agent_names,
                launch_config=launch_config,
            ),
            launch_config,
        )

    def _reload_container(self, container: _DockerContainer) -> None:
        try:
            container.reload()
        except self._docker_errors.DockerException as exc:
            msg = f"Failed to refresh Docker worker state: {exc}"
            raise WorkerBackendError(msg) from exc

    def _wait_for_ready(self, container: _DockerContainer) -> str:
        host_port = self._container_host_port(container)
        if host_port is None:
            msg = "Docker worker is missing a published port."
            raise WorkerBackendError(msg)

        endpoint_root = self._endpoint_root(host_port)
        healthz_url = f"{endpoint_root}/healthz"
        deadline = time.time() + self.config.ready_timeout_seconds
        with httpx.Client(timeout=min(5.0, self.config.ready_timeout_seconds)) as client:
            while True:
                self._reload_container(container)
                if not self._container_is_running(container):
                    msg = "Docker worker stopped before it became ready." + self._container_logs_excerpt(container)
                    raise WorkerBackendError(msg)

                try:
                    response = client.get(healthz_url)
                except httpx.HTTPError:
                    response = None

                if response is not None and 200 <= response.status_code < 300:
                    compatibility_error = _worker_health_compatibility_error(
                        response,
                        image=self.config.image,
                    )
                    if compatibility_error is not None:
                        raise _WorkerImageIncompatibleError(compatibility_error)
                    return f"{endpoint_root}/api/sandbox-runner/execute"

                if time.time() >= deadline:
                    msg = (
                        f"Docker worker did not become ready within {self.config.ready_timeout_seconds:.0f}s."
                        + self._container_logs_excerpt(container)
                    )
                    raise WorkerBackendError(msg)
                time.sleep(_READY_POLL_INTERVAL_SECONDS)

    def _record_failure_locked(
        self,
        paths: LocalWorkerStatePaths,
        metadata: _DockerWorkerMetadata,
        failure_reason: str,
        *,
        now: float,
        stop_container: bool,
    ) -> WorkerHandle:
        container = self._read_container(metadata.container_name)
        if stop_container:
            self._stop_container(container)
        write_lifecycle_state(
            metadata,
            mark_worker_failed(
                read_lifecycle_state(metadata),
                now=now,
                failure_reason=failure_reason,
            ),
        )
        self._save_metadata(paths, metadata)
        return self._to_handle(metadata, container, now=now, paths=paths)

    def _container_env(self, worker_key: str) -> dict[str, str]:
        dedicated_root = Path(self.config.storage_mount_path)
        startup_runtime_paths = self._worker_runtime_paths(
            worker_key=worker_key,
            dedicated_root=dedicated_root,
        )
        # Docker bind-mounts canonical agent state under the same root used as
        # MINDROOM_STORAGE_PATH. Keep the runner's shared-storage root aligned
        # so agent workspaces resolve to mounted host files, not an empty worker
        # state directory.
        shared_storage_root = self.config.storage_mount_path
        env = {
            SANDBOX_RUNTIME_ENV_BY_KEY["runner_mode"]: "true",
            SANDBOX_RUNTIME_ENV_BY_KEY["runner_execution_mode"]: "forkserver",
            _RUNNER_PORT_ENV_NAME: str(self.config.worker_port),
            _STARTUP_RUNTIME_PATHS_ENV: json.dumps(
                serialize_runtime_paths(startup_runtime_paths),
                separators=(",", ":"),
                sort_keys=True,
            ),
            "MINDROOM_STORAGE_PATH": self.config.storage_mount_path,
            _SHARED_STORAGE_ROOT_ENV: shared_storage_root,
            SHARED_CREDENTIALS_PATH_ENV: f"{self.config.storage_mount_path}/.shared_credentials",
            _DEDICATED_WORKER_KEY_ENV: worker_key,
            _DEDICATED_WORKER_ROOT_ENV: self.config.storage_mount_path,
            "HOME": self._container_home_path(worker_key),
            _TOKEN_ENV_NAME: self.auth_token,
        }
        if self.config.host_config_path is not None:
            env["MINDROOM_CONFIG_PATH"] = self.config.config_path
        env.update(self.config.extra_env)
        return env

    def _container_env_matches(
        self,
        container: _DockerContainer | None,
        *,
        expected_env: dict[str, str],
    ) -> bool:
        if container is None:
            return False
        attrs = container.attrs
        config = attrs.get("Config")
        if not isinstance(config, dict):
            return False
        raw_env = cast("dict[str, object]", config).get("Env")
        if not isinstance(raw_env, list):
            return False

        actual_env: dict[str, str] = {}
        for item in raw_env:
            if not isinstance(item, str) or "=" not in item:
                continue
            name, value = item.split("=", 1)
            actual_env[name] = value
        return all(actual_env.get(name) == value for name, value in expected_env.items())

    def _worker_runtime_config_path(self) -> Path:
        return Path(self.config.config_path)

    def _worker_runtime_paths(
        self,
        *,
        worker_key: str,
        dedicated_root: Path,
    ) -> RuntimePaths:
        return build_dedicated_worker_runtime_paths(
            runtime_paths=self._runtime_paths,
            backend_name="Docker",
            worker_key=worker_key,
            config_path=self._worker_runtime_config_path(),
            dedicated_root=dedicated_root,
            worker_port=self.config.worker_port,
            shared_storage_root=self.config.storage_mount_path,
            extra_env=self.config.extra_env,
        )

    def _container_home_path(self, worker_key: str) -> str:
        agent_name = worker_key_agent_name(worker_key)
        if agent_name is None:
            return self.config.storage_mount_path
        if resolved_worker_key_scope(worker_key) == "user_agent":
            # Private workspaces can be renamed in config, so the request
            # preparation layer remains the source of truth for the command cwd.
            return self.config.storage_mount_path
        return str(Path(self.config.storage_mount_path) / "agents" / agent_name / "workspace")

    def _container_volumes(
        self,
        paths: LocalWorkerStatePaths,
        *,
        worker_key: str | None = None,
        private_agent_names: frozenset[str] | None = None,
    ) -> dict[str, dict[str, str]]:
        volumes = {
            str(paths.root): {"bind": self.config.storage_mount_path, "mode": "rw"},
        }
        if worker_key is not None:
            for host_path, container_path, read_only in self._scoped_storage_mount_specs(
                worker_key,
                private_agent_names=private_agent_names,
            ):
                volumes[str(host_path)] = {
                    "bind": container_path,
                    "mode": "ro" if read_only else "rw",
                }
        mount_specs, _projection = self._projection_manager.config_mount_specs(
            paths,
            worker_key=worker_key,
        )
        for host_path, container_path, read_only in mount_specs:
            volumes[str(host_path)] = {
                "bind": container_path,
                "mode": "ro" if read_only else "rw",
            }
        return volumes

    def _scoped_storage_mount_specs(
        self,
        worker_key: str,
        *,
        private_agent_names: frozenset[str] | None,
    ) -> list[tuple[Path, str, bool]]:
        mount_specs = [
            (planned_root.local_path, str(planned_root.worker_visible_path), False)
            for planned_root in plan_scoped_visible_state_roots(
                worker_key=worker_key,
                local_shared_storage_root=self._storage_path,
                worker_visible_shared_storage_root=Path(self.config.storage_mount_path),
                private_agent_names=private_agent_names,
                allow_unknown_worker_key=False,
                resolved_agent_policies=self._projection_manager.current_resolved_agent_policies(),
            )
        ]
        validate_unique_worker_visible_paths(
            (container_path for _host_path, container_path, _read_only in mount_specs),
            worker_key=worker_key,
            duplicate_label="Docker mount",
        )
        return mount_specs

    def _container_labels(
        self,
        metadata: _DockerWorkerMetadata,
        *,
        launch_config_hash: str,
    ) -> dict[str, str]:
        labels = {
            _LABEL_COMPONENT: _LABEL_COMPONENT_VALUE,
            _LABEL_MANAGED_BY: _LABEL_MANAGED_BY_VALUE,
            _LABEL_NAME: _LABEL_NAME_VALUE,
            _LABEL_WORKER_ID: metadata.worker_id,
            _LABEL_LAUNCH_CONFIG_HASH: launch_config_hash,
            _LABEL_RUNTIME_NAMESPACE: self._runtime_namespace,
        }
        labels.update(self.config.extra_labels)
        return labels

    def _compute_launch_config_hash(self, *, image_identity: str | None = None) -> str:
        resolved_image_identity = image_identity or _resolved_docker_image_identity(
            self.config.image,
            client=self._client,
            docker_errors=self._docker_errors,
        )
        config_payload = {
            "auth_token": self.auth_token or "",
            "config_path": self.config.config_path,
            "config_contents_hash": _host_config_contents_hash(self.config.host_config_path),
            "extra_env": self.config.extra_env,
            "extra_labels": self.config.extra_labels,
            "host_config_path": str(self.config.host_config_path or ""),
            "image": self.config.image,
            "resolved_image": resolved_image_identity,
            "name_prefix": self.config.name_prefix,
            "publish_host": self.config.publish_host,
            "storage_mount_path": self.config.storage_mount_path,
            "workers_root": str(self._workers_root),
            "user": self.config.user or "",
            "worker_port": self.config.worker_port,
        }
        normalized = json.dumps(config_payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

    def _resolve_launch_config(self) -> _DockerLaunchConfig:
        image_identity, image_resolved = _docker_image_identity_state(
            self.config.image,
            client=self._client,
            docker_errors=self._docker_errors,
        )
        return self._launch_config_from_image_identity(
            image_identity,
            image_resolved=image_resolved,
        )

    def _launch_config_from_image_identity(
        self,
        image_identity: str,
        *,
        image_resolved: bool,
    ) -> _DockerLaunchConfig:
        return _DockerLaunchConfig(
            image_reference=image_identity if image_resolved else self.config.image,
            launch_config_hash=self._compute_launch_config_hash(image_identity=image_identity),
        )

    def _container_launch_config_hash(self, container: _DockerContainer | None) -> str | None:
        if container is None:
            return None
        attrs = container.attrs
        config = attrs.get("Config")
        if not isinstance(config, dict):
            return None
        labels = cast("dict[str, object]", config).get("Labels")
        launch_config_hash = (
            cast("dict[str, object]", labels).get(_LABEL_LAUNCH_CONFIG_HASH) if isinstance(labels, dict) else None
        )
        if isinstance(launch_config_hash, str) and launch_config_hash:
            return launch_config_hash
        return None

    def _container_image_identity(self, container: _DockerContainer | None) -> str | None:
        if container is None:
            return None

        attrs = container.attrs
        raw_image = attrs.get("Image")
        if isinstance(raw_image, str) and raw_image.strip():
            return raw_image

        config = attrs.get("Config")
        if not isinstance(config, dict):
            return None
        config_image = cast("dict[str, object]", config).get("Image")
        if isinstance(config_image, str) and config_image.strip():
            return config_image
        return None

    def _compatible_launch_config_hashes(self, container: _DockerContainer | None) -> set[str]:
        current_image_identity, image_resolved = _docker_image_identity_state(
            self.config.image,
            client=self._client,
            docker_errors=self._docker_errors,
        )
        compatible_hashes = {self._compute_launch_config_hash(image_identity=current_image_identity)}
        container_image_identity = self._container_image_identity(container)
        if container_image_identity is None:
            return compatible_hashes

        if not image_resolved:
            compatible_hashes.add(self._compute_launch_config_hash(image_identity=container_image_identity))
            return compatible_hashes

        if container_image_identity == current_image_identity:
            compatible_hashes.add(self._compute_launch_config_hash(image_identity=self.config.image))
        return compatible_hashes

    def _container_mount_layout_matches(
        self,
        container: _DockerContainer | None,
        *,
        expected_mounts: list[tuple[Path, str, bool]],
    ) -> bool:
        if container is None:
            return False

        attrs = container.attrs
        mounts = attrs.get("Mounts", [])
        if not isinstance(mounts, list):
            return False

        actual_layout: set[tuple[str, str, bool]] = set()
        for mount in mounts:
            if not isinstance(mount, dict):
                return False
            mount_data = cast("dict[str, object]", mount)
            mount_type = mount_data.get("Type")
            source = mount_data.get("Source")
            destination = mount_data.get("Destination")
            if mount_type != "bind" or not isinstance(source, str) or not isinstance(destination, str):
                continue
            writable = mount_data.get("RW")
            if isinstance(writable, bool):
                read_only = not writable
            else:
                mode = mount_data.get("Mode")
                read_only = "ro" in mode if isinstance(mode, str) else False
            actual_layout.add((str(Path(source).expanduser().resolve()), destination, read_only))

        expected_layout = {
            (str(host_path.expanduser().resolve()), container_path, read_only)
            for host_path, container_path, read_only in expected_mounts
        }
        return actual_layout == expected_layout

    def _container_status(self, container: _DockerContainer) -> str | None:
        status = container.status
        if isinstance(status, str):
            return status
        attrs = container.attrs
        state = attrs.get("State", {})
        if isinstance(state, dict):
            state_status = state.get("Status")
            if isinstance(state_status, str):
                return state_status
        return None

    def _container_logs_excerpt(self, container: _DockerContainer, *, tail: int = 50) -> str:
        """Return a short tail of a worker container's logs for failure diagnostics.

        A worker that exits during startup (for example because its projected
        config fails validation) only records the cause in its own logs, so the
        backend surfaces that tail in the raised error instead of forcing
        operators to inspect the container by hand.
        """
        try:
            raw_logs = container.logs(tail=tail)
        except self._docker_errors.DockerException:
            return ""
        raw_text = raw_logs.decode("utf-8", errors="replace") if isinstance(raw_logs, bytes) else str(raw_logs)
        text = redact_sensitive_text(raw_text.strip(), max_length=_CONTAINER_LOG_EXCERPT_MAX_CHARS)
        if not text:
            return ""
        return f"\nRecent worker container logs:\n{text}"

    def _container_is_running(self, container: _DockerContainer | None) -> bool:
        if container is None:
            return False
        return self._container_status(container) == "running"

    def _container_host_port(self, container: _DockerContainer | None) -> int | None:
        host_port: int | None = None
        if container is None:
            return host_port

        attrs = container.attrs
        network_settings = attrs.get("NetworkSettings")
        if not isinstance(network_settings, dict):
            return host_port
        ports = cast("dict[str, object]", network_settings).get("Ports")
        bindings = (
            cast("dict[str, object]", ports).get(f"{self.config.worker_port}/tcp") if isinstance(ports, dict) else None
        )
        first_binding = bindings[0] if isinstance(bindings, list) and bindings else None
        raw_host_port = (
            cast("dict[str, object]", first_binding).get("HostPort") if isinstance(first_binding, dict) else None
        )
        if not isinstance(raw_host_port, str):
            return host_port
        try:
            host_port = int(raw_host_port)
        except (TypeError, ValueError):
            host_port = None
        return host_port

    def _container_id(self, container: _DockerContainer | None) -> str | None:
        if container is None:
            return None
        container_id = container.id
        return container_id if isinstance(container_id, str) and container_id else None

    def _stop_container(self, container: _DockerContainer | None) -> None:
        if container is None or not self._container_is_running(container):
            return
        try:
            container.stop(timeout=10)
            container.reload()
        except self._docker_errors.DockerException as exc:
            msg = f"Failed to stop Docker worker: {exc}"
            raise WorkerBackendError(msg) from exc

    def _remove_container(self, container: _DockerContainer | None) -> None:
        if container is None:
            return
        try:
            container.remove(force=True)
        except self._docker_errors.NotFound:
            return
        except self._docker_errors.DockerException as exc:
            msg = f"Failed to remove Docker worker: {exc}"
            raise WorkerBackendError(msg) from exc

    def _endpoint_root(self, host_port: int) -> str:
        return f"http://{self.config.endpoint_host}:{host_port}"

    def _endpoint_for_host_port(self, host_port: int | None) -> str:
        if host_port is None:
            return "/api/sandbox-runner/execute"
        return f"{self._endpoint_root(host_port)}/api/sandbox-runner/execute"

    def _effective_status(
        self,
        metadata: _DockerWorkerMetadata,
        container: _DockerContainer | None,
        *,
        now: float,
    ) -> WorkerStatus:
        if metadata.status == "failed":
            return "failed"

        if container is None or not self._container_is_running(container):
            return "idle" if metadata.status != "starting" else "starting"

        if metadata.status == "starting":
            return "starting"
        if now - metadata.last_used_at >= self.idle_timeout_seconds:
            return "idle"
        return "ready"

    def _to_handle(
        self,
        metadata: _DockerWorkerMetadata,
        container: _DockerContainer | None,
        *,
        now: float,
        paths: LocalWorkerStatePaths,
    ) -> WorkerHandle:
        host_port = self._container_host_port(container) or metadata.host_port
        endpoint = self._endpoint_for_host_port(host_port)
        return WorkerHandle(
            worker_id=metadata.worker_id,
            worker_key=metadata.worker_key,
            endpoint=endpoint,
            auth_token=self.auth_token,
            status=self._effective_status(metadata, container, now=now),
            backend_name=self.backend_name,
            last_used_at=metadata.last_used_at,
            created_at=metadata.created_at,
            last_started_at=metadata.last_started_at,
            expires_at=None,
            startup_count=metadata.startup_count,
            failure_count=metadata.failure_count,
            failure_reason=metadata.failure_reason,
            debug_metadata={
                "container_name": metadata.container_name,
                "container_id": self._container_id(container) or metadata.container_id or "",
                "host_port": str(host_port or ""),
                "state_root": str(paths.root),
                "api_root": endpoint.removesuffix("/execute").rstrip("/"),
            },
        )
