"""Tests for the Docker worker backend."""

from __future__ import annotations

import json
import os
import subprocess
import sys
import textwrap
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace

import httpx
import pytest
import yaml

from mindroom.agents import _load_context_files
from mindroom.config.main import load_config
from mindroom.constants import (
    RuntimePaths,
    deserialize_runtime_paths,
    resolve_primary_runtime_paths,
    resolve_runtime_paths,
    runtime_paths_with_storage_root,
)
from mindroom.runtime_env_policy import SHARED_CREDENTIALS_PATH_ENV
from mindroom.tool_system.worker_routing import (
    ToolExecutionIdentity,
    resolve_unscoped_worker_key,
    resolve_worker_key,
    worker_dir_name,
    worker_root_path,
)
from mindroom.workers.backend import WorkerBackendError
from mindroom.workers.backends._dedicated_worker_common import build_dedicated_worker_runtime_paths
from mindroom.workers.backends.docker import (
    DockerWorkerBackend,
    _DockerLaunchConfig,
    _load_docker_client_and_errors,
    _worker_health_compatibility_error,
    _WorkerImageIncompatibleError,
    ensure_docker_dependencies,
)
from mindroom.workers.backends.docker_config import (
    _default_docker_user_for_os,
    _DockerWorkerBackendConfig,
    _read_docker_user,
    docker_backend_config_signature,
)
from mindroom.workers.backends.docker_projection import (
    _PROJECTED_CONFIGS_DIRNAME,
    _WORKER_CONFIG_STATE_DIRNAME,
    DockerProjectionManager,
)
from mindroom.workers.backends.local import local_worker_state_paths_for_root
from mindroom.workers.compatibility import WORKER_PROTOCOL_VERSION
from mindroom.workers.models import WorkerReadyProgress, WorkerSpec
from mindroom.workers.runtime import primary_worker_backend_available, primary_worker_backend_name
from mindroom.workspaces import resolve_agent_workspace_from_state_path

_TEST_AUTH_TOKEN = "test-token"  # noqa: S105
_ROTATED_AUTH_TOKEN = "rotated-token"  # noqa: S105
_TEST_UNSCOPED_WORKER_KEY = "v1:default:unscoped:code"


class _FakeDockerError(Exception):
    pass


class _FakeNotFoundError(_FakeDockerError):
    pass


class _FakeContainer:
    def __init__(
        self,
        *,
        name: str,
        image: str,
        image_identity: str,
        host_port: int,
        environment: dict[str, str],
        labels: dict[str, str],
        user: str | None,
        status: str = "running",
    ) -> None:
        self.name = name
        self.image = image
        self.status = status
        self.id = f"{name}-id"
        self.attrs = {
            "Image": image_identity,
            "State": {"Status": status},
            "Config": {
                "Image": image,
                "Env": [f"{key}={value}" for key, value in sorted(environment.items())],
                "Labels": dict(labels),
                "User": user or "",
            },
            "Mounts": [],
            "NetworkSettings": {
                "Ports": {
                    "8766/tcp": [{"HostIp": "127.0.0.1", "HostPort": str(host_port)}],
                },
            },
        }
        self.started = 0
        self.stopped = 0
        self.removed = 0
        self.logs_output: bytes = b""

    def reload(self) -> None:
        self.attrs["State"]["Status"] = self.status

    def start(self) -> None:
        self.started += 1
        self.status = "running"
        self.reload()

    def stop(self, timeout: int = 10) -> None:
        assert timeout == 10
        self.stopped += 1
        self.status = "exited"
        self.reload()

    def remove(self, force: bool = True) -> None:
        assert force is True
        self.removed += 1
        self.status = "removed"

    def logs(self, *, tail: int = 100) -> bytes:
        assert tail > 0
        return self.logs_output


class _FakeContainersApi:
    def __init__(
        self,
        *,
        images: _FakeImagesApi | None = None,
        auto_pull_missing_image: bool = False,
    ) -> None:
        self.by_name: dict[str, _FakeContainer] = {}
        self.created_containers: list[_FakeContainer] = []
        self.run_calls: list[dict[str, object]] = []
        self.next_host_port = 43001
        self.images = images
        self.auto_pull_missing_image = auto_pull_missing_image

    def get(self, name: str) -> _FakeContainer:
        container = self.by_name.get(name)
        if container is None or container.status == "removed":
            raise _FakeNotFoundError(name)
        return container

    def run(self, image: str, **kwargs: object) -> _FakeContainer:
        if self.auto_pull_missing_image and self.images is not None and image not in self.images.by_name:
            self.images.by_name[image] = _FakeImage("sha256:auto-pulled-image")

        image_identity = image
        if self.images is not None:
            docker_image = self.images.by_name.get(image)
            if docker_image is not None:
                image_identity = docker_image.id

        host_port = self.next_host_port
        self.next_host_port += 1
        environment = kwargs.get("environment")
        labels = kwargs.get("labels")
        volumes = kwargs.get("volumes")
        container = _FakeContainer(
            name=str(kwargs["name"]),
            image=image,
            image_identity=image_identity,
            host_port=host_port,
            environment=dict(environment) if isinstance(environment, dict) else {},
            labels=dict(labels) if isinstance(labels, dict) else {},
            user=str(kwargs["user"]) if kwargs.get("user") is not None else None,
        )
        if isinstance(volumes, dict):
            container.attrs["Mounts"] = [
                {
                    "Type": "bind",
                    "Source": source,
                    "Destination": str(spec.get("bind", "")),
                    "Mode": str(spec.get("mode", "")),
                    "RW": str(spec.get("mode", "rw")) != "ro",
                }
                for source, spec in volumes.items()
                if isinstance(source, str) and isinstance(spec, dict)
            ]
        self.by_name[container.name] = container
        self.created_containers.append(container)
        self.run_calls.append({"image": image, **kwargs})
        return container


class _FakeImage:
    def __init__(self, image_id: str) -> None:
        self.id = image_id


class _FakeImagesApi:
    def __init__(self) -> None:
        self.by_name: dict[str, _FakeImage] = {}
        self.pulls: list[str] = []
        self.pull_image_id: str | None = None

    def get(self, name: str) -> _FakeImage:
        image = self.by_name.get(name)
        if image is None:
            raise _FakeNotFoundError(name)
        return image

    def pull(self, name: str) -> _FakeImage:
        self.pulls.append(name)
        if self.pull_image_id is None:
            msg = f"pull failed: {name}"
            raise _FakeDockerError(msg)
        image = _FakeImage(self.pull_image_id)
        self.by_name[name] = image
        return image


class _FakeDockerClient:
    def __init__(self, *, auto_pull_missing_image: bool = False) -> None:
        self.images = _FakeImagesApi()
        self.containers = _FakeContainersApi(
            images=self.images,
            auto_pull_missing_image=auto_pull_missing_image,
        )


class _WorkerThreadContext(threading.local):
    worker_key: str | None = None


class _ConcurrentTagUpdate:
    def __init__(
        self,
        backend: DockerWorkerBackend,
        *,
        recovering_key: str,
        stale_reader_key: str,
    ) -> None:
        self.backend = backend
        self.recovering_key = recovering_key
        self.stale_reader_key = stale_reader_key
        self.recovering_container_name = backend._container_name_for_worker(recovering_key)
        self.stale_reader_container_name = backend._container_name_for_worker(stale_reader_key)
        self.original_resolve_launch_config = backend._resolve_launch_config
        self.stale_launch_config = self.original_resolve_launch_config()
        self.thread_context = _WorkerThreadContext()
        self.stale_launch_resolved = threading.Event()
        self.recovering_worker_ready = threading.Event()

    def resolve_launch_config(self) -> _DockerLaunchConfig:
        launch_config = self.original_resolve_launch_config()
        if self.thread_context.worker_key == self.stale_reader_key:
            self.stale_launch_resolved.set()
            assert self.recovering_worker_ready.wait(timeout=5.0)
            return self.stale_launch_config
        return launch_config

    def wait_for_ready(self, container: _FakeContainer) -> str:
        if container.attrs["Image"] == "sha256:image-v1":
            message = "stale worker image"
            raise _WorkerImageIncompatibleError(message)
        if self.thread_context.worker_key == self.recovering_key:
            self.recovering_worker_ready.set()
        host_port = self.backend._container_host_port(container)
        return f"http://127.0.0.1:{host_port}/api/sandbox-runner/execute"

    def ensure_worker(self, worker_key: str) -> None:
        self.thread_context.worker_key = worker_key
        self.backend.ensure_worker(WorkerSpec(worker_key), now=0.0)


def _noop_sync_shared_credentials(*_args: object, **_kwargs: object) -> None:
    return None


def _projected_config_fixture(tmp_path: Path) -> tuple[str, dict[str, Path]]:
    plugin_root = tmp_path / "plugins" / "my-plugin"
    plugin_root.mkdir(parents=True)
    (plugin_root / "plugin.py").write_text("PLUGIN_VERSION = 'v1'\n", encoding="utf-8")
    knowledge_root = tmp_path / "knowledge_docs"
    knowledge_root.mkdir()
    (knowledge_root / "guide.md").write_text("# Guide v1\n", encoding="utf-8")
    context_file = tmp_path / "agents" / "code" / "workspace" / "context.md"
    context_file.parent.mkdir(parents=True)
    context_file.write_text("# Context\n", encoding="utf-8")
    memory_root = tmp_path / "memory_files"
    memory_root.mkdir()
    primary_runtime_data = tmp_path / "mindroom_data" / "credentials"
    primary_runtime_data.mkdir(parents=True)
    (primary_runtime_data / "secret.json").write_text('{"api_key":"leak"}', encoding="utf-8")
    return (
        """
plugins:
  - ./plugins/my-plugin
knowledge_bases:
  docs:
    path: ./knowledge_docs
memory:
  backend: file
  file:
    path: ./memory_files
agents:
  code:
    display_name: Code
    role: Test
    model: default
    knowledge_bases: [docs]
    context_files:
      - context.md
models:
  default:
    provider: openai
    id: test-model
    api_key: sk-model-secret
voice:
  enabled: true
  stt:
    provider: openai
    model: whisper-1
    api_key: sk-voice-secret
teams:
  helpers:
    agents: [code]
    mode: collaborate
cultures:
  engineering:
    description: Keep things clean
    agents: [code]
    mode: automatic
authorization:
  global_users:
    - "@owner:example.org"
matrix_room_access:
  mode: single_user_private
mindroom_user:
  username: mindroom
""".lstrip(),
        {
            "plugin_root": plugin_root,
            "knowledge_root": knowledge_root,
            "context_file": context_file,
            "memory_root": memory_root,
        },
    )


def _multi_agent_projected_config_fixture(tmp_path: Path) -> tuple[str, dict[str, Path]]:
    alpha_knowledge_root = tmp_path / "knowledge_alpha"
    alpha_knowledge_root.mkdir()
    (alpha_knowledge_root / "a.txt").write_text("alpha knowledge\n", encoding="utf-8")
    beta_knowledge_root = tmp_path / "knowledge_beta"
    beta_knowledge_root.mkdir()
    (beta_knowledge_root / "b.txt").write_text("beta knowledge\n", encoding="utf-8")
    alpha_context = tmp_path / "agents" / "alpha" / "workspace" / "alpha.md"
    alpha_context.parent.mkdir(parents=True)
    alpha_context.write_text("# Alpha\n", encoding="utf-8")
    beta_context = tmp_path / "agents" / "beta" / "workspace" / "beta.md"
    beta_context.parent.mkdir(parents=True)
    beta_context.write_text("# Beta\n", encoding="utf-8")
    memory_root = tmp_path / "memory_files"
    memory_root.mkdir()
    return (
        """
knowledge_bases:
  a:
    path: ./knowledge_alpha
  b:
    path: ./knowledge_beta
memory:
  backend: file
  file:
    path: ./memory_files
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: shared
    knowledge_bases: [a]
    context_files:
      - alpha.md
  beta:
    display_name: Beta
    role: Beta test
    model: default
    worker_scope: shared
    knowledge_bases: [b]
    context_files:
      - beta.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
        {
            "alpha_context": alpha_context,
            "beta_context": beta_context,
            "alpha_knowledge_root": alpha_knowledge_root,
            "beta_knowledge_root": beta_knowledge_root,
        },
    )


def _private_user_agent_projected_config_fixture(tmp_path: Path) -> tuple[str, dict[str, Path]]:
    alpha_knowledge_root = tmp_path / "knowledge_alpha"
    alpha_knowledge_root.mkdir()
    (alpha_knowledge_root / "a.txt").write_text("alpha knowledge\n", encoding="utf-8")
    beta_knowledge_root = tmp_path / "knowledge_beta"
    beta_knowledge_root.mkdir()
    (beta_knowledge_root / "b.txt").write_text("beta knowledge\n", encoding="utf-8")
    alpha_context = tmp_path / "agents" / "alpha" / "workspace" / "alpha.md"
    alpha_context.parent.mkdir(parents=True)
    alpha_context.write_text("# Alpha\n", encoding="utf-8")
    beta_context = tmp_path / "agents" / "beta" / "workspace" / "beta.md"
    beta_context.parent.mkdir(parents=True)
    beta_context.write_text("# Beta\n", encoding="utf-8")
    return (
        """
knowledge_bases:
  a:
    path: ./knowledge_alpha
  b:
    path: ./knowledge_beta
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    private:
      per: user_agent
    knowledge_bases: [a]
    context_files:
      - alpha.md
  beta:
    display_name: Beta
    role: Beta test
    model: default
    worker_scope: shared
    knowledge_bases: [b]
    context_files:
      - beta.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
        {
            "alpha_context": alpha_context,
            "beta_context": beta_context,
            "alpha_knowledge_root": alpha_knowledge_root,
            "beta_knowledge_root": beta_knowledge_root,
        },
    )


def _knowledge_base_collision_fixture(tmp_path: Path) -> str:
    first_root = tmp_path / "knowledge_collision_a"
    first_root.mkdir()
    (first_root / "a.txt").write_text("first\n", encoding="utf-8")
    second_root = tmp_path / "knowledge_collision_b"
    second_root.mkdir()
    (second_root / "b.txt").write_text("second\n", encoding="utf-8")
    memory_root = tmp_path / "memory_files"
    memory_root.mkdir()
    return """
knowledge_bases:
  "docs/a":
    path: ./knowledge_collision_a
  "docs:a":
    path: ./knowledge_collision_b
memory:
  backend: file
  file:
    path: ./memory_files
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: shared
    knowledge_bases: ["docs/a", "docs:a"]
models:
  default:
    provider: openai
    id: test-model
""".lstrip()


def _projection_root(volumes: dict[str, dict[str, str]]) -> Path:
    return next(Path(source) for source, spec in volumes.items() if spec["bind"] == "/app/config-host")


def _assert_projected_worker_mounts(
    tmp_path: Path,
    volumes: dict[str, dict[str, str]],
    projected_paths: dict[str, Path],
) -> Path:
    worker_root = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY)
    state_root = str(worker_root)
    assert volumes[state_root]["bind"] == "/app/worker"

    projection_root = _projection_root(volumes)
    assert projection_root.parent == (
        worker_root_path(tmp_path, "__mindroom_root__").parent
        / _PROJECTED_CONFIGS_DIRNAME
        / worker_dir_name(_TEST_UNSCOPED_WORKER_KEY)
    )
    assert worker_root not in projection_root.parents
    assert str(tmp_path) not in volumes
    assert str(tmp_path / "mindroom_data") not in volumes
    assert str(projected_paths["plugin_root"].resolve()) not in volumes
    assert str(projected_paths["knowledge_root"].resolve()) not in volumes
    assert str(projected_paths["context_file"].resolve()) not in volumes
    assert str(projected_paths["memory_root"].resolve()) not in volumes
    return projection_root


def _assert_projected_config_snapshot(projection_root: Path, tmp_path: Path) -> None:
    projected_config = (projection_root / "config.yaml").read_text(encoding="utf-8")
    assert "plugins:\n- ./.mindroom-worker-assets/plugins/00-my-plugin" in projected_config
    assert "path: ./.mindroom-worker-assets/knowledge_bases/docs" in projected_config
    assert f"path: /app/worker/{_WORKER_CONFIG_STATE_DIRNAME}/memory/file" in projected_config
    assert "- ./.mindroom-worker-assets/agents/code/context_files/00-context.md" in projected_config

    projected_context_path = (
        projection_root / ".mindroom-worker-assets" / "agents" / "code" / "context_files" / "00-context.md"
    )
    assert projected_context_path.read_text(encoding="utf-8") == "# Context\n"
    projected_plugin_path = projection_root / ".mindroom-worker-assets" / "plugins" / "00-my-plugin" / "plugin.py"
    assert projected_plugin_path.read_text(encoding="utf-8") == "PLUGIN_VERSION = 'v1'\n"
    projected_knowledge_path = projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "docs" / "guide.md"
    assert projected_knowledge_path.read_text(encoding="utf-8") == "# Guide v1\n"
    assert (projection_root / ".env").read_text(encoding="utf-8") == ""
    assert (projection_root / ".projection-ready").read_text(encoding="utf-8") == "ready\n"
    assert (
        worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / _WORKER_CONFIG_STATE_DIRNAME / "memory" / "file"
    ).is_dir()


def test_docker_worker_projection_rewrites_mapping_plugin_paths(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Mapping-style plugin entries must be copied into the worker projection."""
    plugin_root = tmp_path / "plugins" / "mapped-plugin"
    plugin_root.mkdir(parents=True)
    (plugin_root / "plugin.py").write_text("PLUGIN_VERSION = 'mapped'\n", encoding="utf-8")
    config_text = """
plugins:
  - path: ./plugins/mapped-plugin
    enabled: true
agents: {}
models:
  default:
    provider: openai
    id: test-model
router:
  model: default
""".lstrip()

    backend, _fake_client, _credential_sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
    )
    projection = backend._projection_manager._projected_config(
        local_worker_state_paths_for_root(tmp_path / "workers" / "mapped-plugin"),
        materialize=True,
    )

    projected_config = (projection.root / "config.yaml").read_text(encoding="utf-8")
    assert "path: ./.mindroom-worker-assets/plugins/00-mapped-plugin" in projected_config
    assert "enabled: true" in projected_config
    projected_plugin_path = projection.root / ".mindroom-worker-assets" / "plugins" / "00-mapped-plugin" / "plugin.py"
    assert projected_plugin_path.read_text(encoding="utf-8") == "PLUGIN_VERSION = 'mapped'\n"


def test_docker_worker_projection_resolves_config_includes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A split host config projects as the fully merged config for the worker."""
    (tmp_path / "agents.yaml").write_text(
        "code:\n  display_name: Code\n  role: Writes code\n",
        encoding="utf-8",
    )
    config_text = (
        "agents: !include agents.yaml\n"
        "models:\n  default:\n    provider: openai\n    id: test-model\n"
        "router:\n  model: default\n"
    )

    backend, _fake_client, _credential_sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
    )
    projection = backend._projection_manager._projected_config(
        local_worker_state_paths_for_root(tmp_path / "workers" / "split-config"),
        materialize=True,
    )

    projected = yaml.safe_load((projection.root / "config.yaml").read_text(encoding="utf-8"))
    assert projected["agents"]["code"]["display_name"] == "Code"


def _backend(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    *,
    idle_timeout_seconds: float = 60.0,
    config_text: str = "agents: {}\n",
    runtime_paths: RuntimePaths | None = None,
    storage_path: Path | None = None,
    host_config_path: Path | None = None,
) -> tuple[DockerWorkerBackend, _FakeDockerClient, list[tuple[str, frozenset[str]]]]:
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=tmp_path / "config.yaml" if host_config_path is None else host_config_path,
        idle_timeout_seconds=idle_timeout_seconds,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={"EXTRA_ENV": "present"},
        extra_labels={"mindroom.ai/tenant": "test"},
    )
    assert config.host_config_path is not None
    config.host_config_path.parent.mkdir(parents=True, exist_ok=True)
    config.host_config_path.write_text(config_text, encoding="utf-8")
    fake_client = _FakeDockerClient()
    fake_client.images.by_name[config.image] = _FakeImage("sha256:image-v1")
    sync_calls: list[tuple[str, frozenset[str]]] = []

    monkeypatch.setattr(
        "mindroom.workers.backends.docker._load_docker_client_and_errors",
        lambda *_args, **_kwargs: (
            fake_client,
            SimpleNamespace(
                DockerException=_FakeDockerError,
                NotFound=_FakeNotFoundError,
            ),
        ),
    )

    def _record_sync_call(
        worker_key: str,
        allowed_services: frozenset[str] = frozenset(),
        **_kwargs: object,
    ) -> None:
        sync_calls.append((worker_key, allowed_services))

    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _record_sync_call,
    )
    backend = DockerWorkerBackend(
        config=config,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=tmp_path if storage_path is None else storage_path,
        runtime_paths=runtime_paths,
    )
    monkeypatch.setattr(
        backend,
        "_wait_for_ready",
        lambda container: f"http://127.0.0.1:{backend._container_host_port(container)}/api/sandbox-runner/execute",
    )
    return backend, fake_client, sync_calls


def _use_real_wait_for_ready(monkeypatch: pytest.MonkeyPatch, backend: DockerWorkerBackend) -> None:
    monkeypatch.setattr(
        backend,
        "_wait_for_ready",
        DockerWorkerBackend._wait_for_ready.__get__(backend),
    )


def _install_health_by_image(
    monkeypatch: pytest.MonkeyPatch,
    fake_client: _FakeDockerClient,
) -> None:
    class _HealthByImage:
        def __init__(self, *, timeout: float) -> None:
            assert timeout > 0

        def __enter__(self) -> _HealthByImage:
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def get(self, url: str) -> httpx.Response:
            port = url.split(":")[2].split("/", maxsplit=1)[0]
            container = next(
                (
                    item
                    for item in fake_client.containers.by_name.values()
                    if item.status == "running"
                    and item.attrs["NetworkSettings"]["Ports"]["8766/tcp"][0]["HostPort"] == port
                ),
                None,
            )
            assert container is not None, f"no running fake container publishes host port {port}"
            protocol = WORKER_PROTOCOL_VERSION if container.attrs["Image"] == "sha256:image-v2" else 0
            return httpx.Response(
                200,
                json={"status": "ok", "mindroom_version": "test", "worker_protocol": protocol},
            )

    monkeypatch.setattr("mindroom.workers.backends.docker.httpx.Client", _HealthByImage)


def _install_stale_health(monkeypatch: pytest.MonkeyPatch) -> None:
    class _StaleHealthClient:
        def __init__(self, *, timeout: float) -> None:
            assert timeout > 0

        def __enter__(self) -> _StaleHealthClient:
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def get(self, _url: str) -> httpx.Response:
            return httpx.Response(
                200,
                json={"status": "ok", "mindroom_version": "test", "worker_protocol": 0},
            )

    monkeypatch.setattr("mindroom.workers.backends.docker.httpx.Client", _StaleHealthClient)


def test_primary_worker_backend_available_uses_runtime_env_values(tmp_path: Path) -> None:
    """Docker backend availability should honor the explicit runtime context."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (config_dir / ".env").write_text(
        (
            "MINDROOM_WORKER_BACKEND=docker\n"
            "MINDROOM_DOCKER_WORKER_IMAGE=test-image\n"
            "MINDROOM_SANDBOX_PROXY_TOKEN=test-token\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    assert primary_worker_backend_name(runtime_paths) == "docker"
    assert runtime_paths.env_value("MINDROOM_DOCKER_WORKER_IMAGE") == "test-image"
    assert primary_worker_backend_available(
        runtime_paths,
        proxy_url=None,
        proxy_token=runtime_paths.env_value("MINDROOM_SANDBOX_PROXY_TOKEN"),
    )


def test_docker_worker_host_config_path_resolves_relative_to_runtime_config_dir(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Relative Docker host-config paths should resolve against the runtime config directory, not cwd."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir()
    config_path = config_dir / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    host_config_path = config_dir / "docker-host-config.yaml"
    host_config_path.write_text("agents: {}\n", encoding="utf-8")

    other_cwd = tmp_path / "other"
    other_cwd.mkdir()
    monkeypatch.chdir(other_cwd)

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_HOST_CONFIG_PATH": "./docker-host-config.yaml",
        },
    )

    config = _DockerWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.host_config_path == host_config_path.resolve()


def test_docker_worker_host_config_path_rejects_directories(tmp_path: Path) -> None:
    """Docker host-config path must point at a file when explicitly configured."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    host_config_dir = tmp_path / "host-config"
    host_config_dir.mkdir()
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_HOST_CONFIG_PATH": str(host_config_dir),
        },
    )

    with pytest.raises(WorkerBackendError, match="points to a directory"):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


def test_runtime_paths_with_storage_root_updates_primary_path_env(tmp_path: Path) -> None:
    """Rebased primary runtimes should keep their exported path contract internally consistent."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "runtime-a",
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
        },
    )

    rebased = runtime_paths_with_storage_root(runtime_paths, tmp_path / "runtime-b")

    assert rebased.storage_root == (tmp_path / "runtime-b").resolve()
    assert rebased.process_env["MINDROOM_CONFIG_PATH"] == str(config_path.resolve())
    assert rebased.process_env["MINDROOM_STORAGE_PATH"] == str((tmp_path / "runtime-b").resolve())


def test_docker_backend_config_signature_ignores_original_runtime_storage_root_when_explicit_storage_path_matches(
    tmp_path: Path,
) -> None:
    """The Docker manager signature should depend on the effective backend storage root, not stale callers."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    env = {
        "MINDROOM_WORKER_BACKEND": "docker",
        "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
    }

    first_runtime = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "runtime-a",
        process_env=env,
    )
    second_runtime = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "runtime-b",
        process_env=env,
    )

    shared_storage_path = tmp_path / "docker-storage"
    assert docker_backend_config_signature(
        first_runtime,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=shared_storage_path,
    ) == docker_backend_config_signature(
        second_runtime,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=shared_storage_path,
    )


def test_docker_backend_from_runtime_reanchors_host_config_projection_and_runtime_env(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Host-config overrides should also re-anchor relative assets and sibling .env values."""
    runtime_dir = tmp_path / "runtime"
    runtime_dir.mkdir()
    runtime_config_path = runtime_dir / "config.yaml"
    runtime_config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: test-model\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (runtime_dir / ".env").write_text("BROWSER_EXECUTABLE_PATH=/runtime/browser\n", encoding="utf-8")

    host_config_dir = tmp_path / "host-config"
    host_config_dir.mkdir()
    host_config_path = host_config_dir / "docker-host-config.yaml"
    host_config_path.write_text(
        "plugins:\n  - ./plugin.py\nmodels:\n  default:\n    provider: openai\n    id: test-model\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (host_config_dir / "plugin.py").write_text("PLUGIN = 1\n", encoding="utf-8")
    (host_config_dir / ".env").write_text("BROWSER_EXECUTABLE_PATH=/hostcfg/browser\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=runtime_config_path,
        storage_path=tmp_path / "storage",
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_HOST_CONFIG_PATH": str(host_config_path),
        },
    )
    config = _DockerWorkerBackendConfig.from_runtime(runtime_paths)
    fake_client = _FakeDockerClient()
    fake_client.images.by_name[config.image] = _FakeImage("sha256:image-v1")
    monkeypatch.setattr(
        "mindroom.workers.backends.docker._load_docker_client_and_errors",
        lambda *_args, **_kwargs: (
            fake_client,
            SimpleNamespace(
                DockerException=_FakeDockerError,
                NotFound=_FakeNotFoundError,
            ),
        ),
    )
    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _noop_sync_shared_credentials,
    )

    backend = DockerWorkerBackend(
        config=config,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=tmp_path / "storage",
        runtime_paths=runtime_paths,
    )
    projection = backend._projection_manager._projected_config(
        local_worker_state_paths_for_root(tmp_path / "workers" / "projection-test"),
        materialize=False,
    )

    assert backend._runtime_paths.config_path == host_config_path.resolve()
    assert backend._runtime_paths.env_value("BROWSER_EXECUTABLE_PATH") == "/hostcfg/browser"
    assert ".mindroom-worker-assets/plugins/00-plugin.py" in projection.projected_yaml
    assert [asset.relative_path.as_posix() for asset in projection.assets] == [
        ".mindroom-worker-assets/plugins/00-plugin.py",
    ]


def test_docker_backend_config_signature_supports_nested_relative_host_config_paths(tmp_path: Path) -> None:
    """Nested relative host-config paths should resolve once and remain usable during signature generation."""
    config_dir = tmp_path / "cfg"
    nested_dir = config_dir / "nested"
    nested_dir.mkdir(parents=True)
    config_path = config_dir / "config.yaml"
    host_config_path = nested_dir / "docker-host-config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    host_config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_HOST_CONFIG_PATH": "./nested/docker-host-config.yaml",
        },
    )

    assert _DockerWorkerBackendConfig.from_runtime(runtime_paths).host_config_path == host_config_path.resolve()
    signature = docker_backend_config_signature(
        runtime_paths,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=tmp_path / "storage",
    )

    assert signature[0] == "docker"


def test_docker_backend_config_signature_keeps_primary_runtime_env_when_reanchoring_host_config(
    tmp_path: Path,
) -> None:
    """Host-config rebasing should not drop Docker backend env that only exists in the primary runtime .env."""
    runtime_dir = tmp_path / "runtime"
    runtime_dir.mkdir()
    hostcfg_dir = tmp_path / "hostcfg"
    hostcfg_dir.mkdir()
    runtime_config_path = runtime_dir / "config.yaml"
    runtime_config_path.write_text("agents: {}\n", encoding="utf-8")
    (runtime_dir / ".env").write_text("MINDROOM_DOCKER_WORKER_IMAGE=image-from-runtime-env\n", encoding="utf-8")
    host_config_path = hostcfg_dir / "docker-host-config.yaml"
    host_config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=runtime_config_path,
        storage_path=tmp_path / "storage",
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_HOST_CONFIG_PATH": str(host_config_path),
        },
    )

    signature = docker_backend_config_signature(
        runtime_paths,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=tmp_path / "storage",
    )

    assert signature[0] == "docker"
    assert primary_worker_backend_available(runtime_paths, proxy_url=None, proxy_token=_TEST_AUTH_TOKEN)


def test_docker_worker_config_rejects_wildcard_endpoint_host_from_publish_host(tmp_path: Path) -> None:
    """Wildcard bind hosts must not become the client-facing worker endpoint host."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_PUBLISH_HOST": "0.0.0.0",  # noqa: S104 - wildcard bind is the scenario under test
        },
    )

    with pytest.raises(WorkerBackendError, match=r"endpoint_host cannot be 0\.0\.0\.0"):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


def test_docker_worker_config_allows_explicit_endpoint_host_with_wildcard_publish_host(tmp_path: Path) -> None:
    """Operators may bind broadly as long as the returned worker endpoint is explicit and reachable."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_PUBLISH_HOST": "0.0.0.0",  # noqa: S104 - wildcard bind is the scenario under test
            "MINDROOM_DOCKER_WORKER_ENDPOINT_HOST": "127.0.0.1",
        },
    )

    config = _DockerWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.publish_host == "0.0.0.0"  # noqa: S104 - wildcard bind is the scenario under test
    assert config.endpoint_host == "127.0.0.1"


@pytest.mark.parametrize(
    ("label_name"),
    [
        "mindroom.ai/launch-config-hash",
        "mindroom.ai/worker-key",
        "mindroom.ai/component",
    ],
)
def test_docker_worker_backend_rejects_reserved_extra_labels(
    tmp_path: Path,
    label_name: str,
) -> None:
    """Reserved Docker labels must fail fast instead of overriding backend-owned metadata."""
    with pytest.raises(WorkerBackendError, match="extra labels cannot override reserved labels"):
        _DockerWorkerBackendConfig(
            image="ghcr.io/mindroom-ai/mindroom:latest",
            worker_port=8766,
            storage_mount_path="/app/worker",
            config_path="/app/config-host/config.yaml",
            host_config_path=tmp_path / "config.yaml",
            idle_timeout_seconds=60.0,
            ready_timeout_seconds=5.0,
            name_prefix="mindroom-worker",
            publish_host="127.0.0.1",
            endpoint_host="127.0.0.1",
            user="1000:1000",
            extra_env={},
            extra_labels={label_name: "user-value"},
        )


@pytest.mark.parametrize(
    ("label_name"),
    [
        "mindroom.ai/launch-config-hash",
        "mindroom.ai/worker-key",
        "mindroom.ai/component",
    ],
)
def test_docker_worker_config_rejects_reserved_extra_labels_from_env(
    tmp_path: Path,
    label_name: str,
) -> None:
    """Runtime env loading should reject reserved Docker labels before backend creation."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_LABELS_JSON": json.dumps({label_name: "user-value"}),
        },
    )

    with pytest.raises(WorkerBackendError, match="extra labels cannot override reserved labels"):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


def test_docker_worker_config_rejects_malformed_json_mapping_env(tmp_path: Path) -> None:
    """Malformed Docker JSON env values should fail closed instead of silently becoming empty mappings."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_ENV_JSON": "{not-json",
        },
    )

    with pytest.raises(WorkerBackendError, match="MINDROOM_DOCKER_WORKER_ENV_JSON must contain a JSON object"):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


@pytest.mark.parametrize(
    ("storage_mount_path", "config_path"),
    [
        ("/app/worker", "/app/worker/config.yaml"),
        ("/app/worker", "/app/worker/nested/config.yaml"),
        ("/app/worker/config-host", "/app/worker/config.yaml"),
    ],
)
def test_docker_worker_backend_rejects_overlapping_config_mount_targets(
    tmp_path: Path,
    storage_mount_path: str,
    config_path: str,
) -> None:
    """Projected config mounts must stay disjoint from the writable worker state root."""
    with pytest.raises(WorkerBackendError, match="config_path must mount outside the worker storage root"):
        _DockerWorkerBackendConfig(
            image="ghcr.io/mindroom-ai/mindroom:latest",
            worker_port=8766,
            storage_mount_path=storage_mount_path,
            config_path=config_path,
            host_config_path=tmp_path / "config.yaml",
            idle_timeout_seconds=60.0,
            ready_timeout_seconds=5.0,
            name_prefix="mindroom-worker",
            publish_host="127.0.0.1",
            endpoint_host="127.0.0.1",
            user="1000:1000",
            extra_env={},
            extra_labels={},
        )


@pytest.mark.parametrize(
    ("storage_mount_path", "config_path"),
    [
        ("/app/worker", "/app/worker/config.yaml"),
        ("/app/worker", "/app/worker/nested/config.yaml"),
        ("/app/worker/config-host", "/app/worker/config.yaml"),
    ],
)
def test_docker_worker_config_rejects_overlapping_config_mount_targets_from_env(
    tmp_path: Path,
    storage_mount_path: str,
    config_path: str,
) -> None:
    """Runtime env loading should reject overlapping Docker config and worker-state mount targets."""
    runtime_config_path = tmp_path / "config.yaml"
    runtime_config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=runtime_config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_STORAGE_MOUNT_PATH": storage_mount_path,
            "MINDROOM_DOCKER_WORKER_CONFIG_PATH": config_path,
        },
    )

    with pytest.raises(WorkerBackendError, match="config_path must mount outside the worker storage root"):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


def _projection_signature_for_hash_seed(hash_seed: str, workspace_root: Path) -> dict[str, object]:
    repo_root = Path(__file__).resolve().parents[1]
    existing_pythonpath = os.environ.get("PYTHONPATH")
    pythonpath_entries = [str(repo_root / "src")]
    if existing_pythonpath:
        pythonpath_entries.append(existing_pythonpath)
    script = (
        textwrap.dedent(
            """
        import json
        from pathlib import Path

        import yaml

        from mindroom.constants import resolve_runtime_paths
        from mindroom.workers.backends.docker_config import _DockerWorkerBackendConfig
        from mindroom.workers.backends.docker_projection import DockerProjectionManager
        from mindroom.workers.backends.local import local_worker_state_paths_for_root

        tmp_path = Path(__WORKSPACE_ROOT__)
        runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config.yaml", storage_path=tmp_path)
        config = _DockerWorkerBackendConfig(
            image="ghcr.io/mindroom-ai/mindroom:latest",
            worker_port=8766,
            storage_mount_path="/app/worker",
            config_path="/app/config-host/config.yaml",
            host_config_path=tmp_path / "config.yaml",
            idle_timeout_seconds=60.0,
            ready_timeout_seconds=5.0,
            name_prefix="mindroom-worker",
            publish_host="127.0.0.1",
            endpoint_host="127.0.0.1",
            user="1000:1000",
            extra_env={},
            extra_labels={},
        )
        manager = DockerProjectionManager(
            config=config,
            projected_configs_root=tmp_path / "projections",
            runtime_paths=runtime_paths,
        )
        paths = local_worker_state_paths_for_root(tmp_path / "workers" / "worker-a")
        projection = manager._projected_config(
            paths,
            worker_key="v1:default:shared:alpha",
            materialize=False,
        )
        projected_config = yaml.safe_load(projection.projected_yaml)
        print(
            json.dumps(
                {
                    "projection_root": projection.root.name,
                    "knowledge_bases": list(projected_config["knowledge_bases"]),
                },
            ),
        )
        """,
        )
        .lstrip()
        .replace("__WORKSPACE_ROOT__", json.dumps(str(workspace_root)))
    )
    completed = subprocess.run(
        [sys.executable, "-c", script],
        check=True,
        capture_output=True,
        cwd=repo_root,
        env={
            **os.environ,
            "PYTHONHASHSEED": hash_seed,
            "PYTHONPATH": os.pathsep.join(pythonpath_entries),
        },
        text=True,
    )
    return json.loads(completed.stdout)


def test_docker_backend_ensures_worker_container_and_bind_mount(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Ensuring one worker should project only configured config assets into the container."""
    config_text, projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    assert handle.worker_key == _TEST_UNSCOPED_WORKER_KEY
    assert handle.backend_name == "docker"
    assert handle.status == "ready"
    assert handle.endpoint.endswith("/api/sandbox-runner/execute")
    assert handle.debug_metadata["state_root"] == str(worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY))
    assert sync_calls == [(_TEST_UNSCOPED_WORKER_KEY, frozenset())]

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    assert env["MINDROOM_SANDBOX_RUNNER_MODE"] == "true"
    assert env["MINDROOM_SANDBOX_RUNNER_EXECUTION_MODE"] == "forkserver"
    assert env["MINDROOM_SANDBOX_PROXY_TOKEN"] == _TEST_AUTH_TOKEN
    assert env["MINDROOM_SANDBOX_DEDICATED_WORKER_KEY"] == _TEST_UNSCOPED_WORKER_KEY
    assert env["MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT"] == "/app/worker"
    assert env["MINDROOM_SANDBOX_SHARED_STORAGE_ROOT"] == "/app/worker"
    assert env["MINDROOM_SHARED_CREDENTIALS_PATH"] == "/app/worker/.shared_credentials"
    assert env["MINDROOM_CONFIG_PATH"] == "/app/config-host/config.yaml"
    assert env["HOME"] == "/app/worker/agents/code/workspace"
    assert env["EXTRA_ENV"] == "present"

    volumes = run_call["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _assert_projected_worker_mounts(tmp_path, volumes, projected_paths)
    _assert_projected_config_snapshot(projection_root, tmp_path)
    assert run_call["user"] == "1000:1000"
    assert run_call["ports"] == {"8766/tcp": ("127.0.0.1", None)}

    labels = run_call["labels"]
    assert isinstance(labels, dict)
    assert labels["mindroom.ai/component"] == "worker"
    assert "mindroom.ai/worker-key" not in labels
    assert labels["mindroom.ai/runtime-namespace"]
    assert labels["mindroom.ai/tenant"] == "test"

    metadata_path = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / "metadata" / "worker.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["status"] == "ready"
    assert metadata["startup_count"] == 1


def test_docker_backend_accepts_manager_progress_sink(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The backend should be able to receive progress sinks during ensure."""
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    progress_events: list[WorkerReadyProgress] = []

    handle = backend.ensure_worker(
        WorkerSpec(_TEST_UNSCOPED_WORKER_KEY),
        now=10.0,
        progress_sink=progress_events.append,
    )

    assert handle.worker_key == _TEST_UNSCOPED_WORKER_KEY
    assert handle.status == "ready"
    assert [event.phase for event in progress_events] == ["cold_start", "ready"]


def test_docker_backend_projects_assets_from_runtime_storage_root(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected Docker assets should resolve ${MINDROOM_STORAGE_PATH} from the active runtime."""
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    plugin_root = runtime_storage / "plugins" / "runtime-plugin"
    plugin_root.mkdir(parents=True)
    (plugin_root / "plugin.py").write_text("PLUGIN_VERSION = 'runtime'\n", encoding="utf-8")
    knowledge_root = runtime_storage / "knowledge_docs"
    knowledge_root.mkdir()
    (knowledge_root / "guide.md").write_text("# Runtime Guide\n", encoding="utf-8")
    context_file = runtime_storage / "agents" / "code" / "workspace" / "context.md"
    context_file.parent.mkdir(parents=True, exist_ok=True)
    context_file.write_text("# Runtime Context\n", encoding="utf-8")

    config_path = tmp_path / "config.yaml"
    runtime_paths = resolve_runtime_paths(config_path=config_path, storage_path=runtime_storage)
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
plugins:
  - ${MINDROOM_STORAGE_PATH}/plugins/runtime-plugin
knowledge_bases:
  docs:
    path: ${MINDROOM_STORAGE_PATH}/knowledge_docs
agents:
  code:
    display_name: Code
    role: Test
    model: default
    knowledge_bases: [docs]
    context_files:
      - context.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = (projection_root / "config.yaml").read_text(encoding="utf-8")

    assert "plugins:\n- ./.mindroom-worker-assets/plugins/00-runtime-plugin" in projected_config
    assert "path: ./.mindroom-worker-assets/knowledge_bases/docs" in projected_config
    assert "- ./.mindroom-worker-assets/agents/code/context_files/00-context.md" in projected_config
    assert (projection_root / ".mindroom-worker-assets" / "plugins" / "00-runtime-plugin" / "plugin.py").read_text(
        encoding="utf-8",
    ) == "PLUGIN_VERSION = 'runtime'\n"
    assert (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "docs" / "guide.md").read_text(
        encoding="utf-8",
    ) == "# Runtime Guide\n"
    assert (
        projection_root / ".mindroom-worker-assets" / "agents" / "code" / "context_files" / "00-context.md"
    ).read_text(encoding="utf-8") == "# Runtime Context\n"


def test_docker_backend_rejects_symlinked_projected_directory_assets(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected directory assets must fail closed instead of dereferencing symlinked children."""
    knowledge_root = tmp_path / "knowledge_docs"
    knowledge_root.mkdir()
    (knowledge_root / "guide.md").write_text("# Guide\n", encoding="utf-8")
    secret_path = tmp_path / ".env"
    secret_path.write_text("TOP_SECRET=shh\n", encoding="utf-8")
    (knowledge_root / "leak.env").symlink_to(secret_path)
    backend, _fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
knowledge_bases:
  docs:
    path: ./knowledge_docs
agents:
  code:
    display_name: Code
    role: Test
    model: default
    knowledge_bases: [docs]
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )

    with pytest.raises(WorkerBackendError, match="Docker worker asset must not contain symlinks"):
        backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)


def test_docker_backend_rejects_symlinked_projected_file_assets(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected file assets must reject symlink roots instead of escaping the agent workspace."""
    secret_path = tmp_path / "secret.md"
    secret_path.write_text("secret\n", encoding="utf-8")
    context_path = tmp_path / "agents" / "code" / "workspace" / "context.md"
    context_path.parent.mkdir(parents=True)
    context_path.symlink_to(secret_path)
    backend, _fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
agents:
  code:
    display_name: Code
    role: Test
    model: default
    context_files:
      - context.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )

    with pytest.raises(WorkerBackendError, match="Agent-owned paths must stay within"):
        backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)


def test_docker_backend_syncs_shared_credentials_from_runtime_storage_root(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Shared-credential mirroring should use the active runtime storage root."""
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    synced_storage_roots: list[Path | None] = []

    def _capture_runtime_storage_root(
        _worker_key: str,
        **kwargs: object,
    ) -> None:
        credentials_manager = kwargs.get("credentials_manager")
        synced_storage_roots.append(
            None if credentials_manager is None else credentials_manager.storage_root,
        )

    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _capture_runtime_storage_root,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    assert synced_storage_roots == [tmp_path.resolve()]


def test_docker_backend_syncs_shared_credentials_from_runtime_shared_path(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Shared-credential mirroring should honor explicit runtime mirror paths."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    shared_credentials_path = (tmp_path / "shared-credentials").resolve()
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={SHARED_CREDENTIALS_PATH_ENV: str(shared_credentials_path)},
    )
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path, runtime_paths=runtime_paths)
    synced_shared_paths: list[Path | None] = []

    def _capture_runtime_shared_path(
        _worker_key: str,
        **kwargs: object,
    ) -> None:
        credentials_manager = kwargs.get("credentials_manager")
        synced_shared_paths.append(
            None if credentials_manager is None else credentials_manager.shared_base_path,
        )

    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _capture_runtime_shared_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    assert synced_shared_paths == [shared_credentials_path]


def test_docker_backend_commits_parent_runtime_env_into_worker_payload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated Docker workers should receive the committed public startup runtime only."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    credentials_path = tmp_path / "google-credentials.json"
    credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    (config_dir / ".env").write_text(
        (
            "MINDROOM_NAMESPACE=alpha1234\n"
            "MATRIX_HOMESERVER=http://dotenv-hs\n"
            "MATRIX_SERVER_NAME=alpha.example\n"
            "BROWSER_EXECUTABLE_PATH=/usr/bin/chromium\n"
            f"GOOGLE_APPLICATION_CREDENTIALS={credentials_path}\n"
            "GOOGLE_CLOUD_PROJECT=demo-project\n"
            "GOOGLE_CLOUD_LOCATION=us-central1\n"
            "ANTHROPIC_API_KEY=sk-secret\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=runtime_storage,
        process_env={
            "MINDROOM_SANDBOX_PROXY_TOKEN": "test-token",
            "MINDROOM_LOCAL_CLIENT_SECRET": "client-secret",
        },
    )
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    local_credentials_path = (
        worker_root_path(runtime_storage, _TEST_UNSCOPED_WORKER_KEY) / ".runtime" / credentials_path.name
    )

    assert env["MINDROOM_CONFIG_PATH"] == "/app/config-host/config.yaml"
    assert committed_runtime.config_path == Path("/app/config-host/config.yaml")
    assert committed_runtime.env_value("MINDROOM_NAMESPACE") == "alpha1234"
    assert committed_runtime.env_value("MATRIX_HOMESERVER") == "http://dotenv-hs"
    assert committed_runtime.env_value("MATRIX_SERVER_NAME") == "alpha.example"
    assert committed_runtime.env_value("BROWSER_EXECUTABLE_PATH") == "/usr/bin/chromium"
    assert committed_runtime.env_value("GOOGLE_APPLICATION_CREDENTIALS") is None
    assert committed_runtime.env_value("GOOGLE_CLOUD_PROJECT") == "demo-project"
    assert committed_runtime.env_value("GOOGLE_CLOUD_LOCATION") == "us-central1"
    assert committed_runtime.env_value("ANTHROPIC_API_KEY") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_PROXY_TOKEN") is None
    assert committed_runtime.env_value("MINDROOM_LOCAL_CLIENT_SECRET") is None
    assert not local_credentials_path.exists()


def test_docker_backend_excludes_internal_file_secrets_from_worker_payload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated Docker workers must not copy control-plane or GitHub file secrets into worker startup env."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    secret_path = tmp_path / "control-secret.txt"
    secret_path.write_text("supersecret\n", encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=runtime_storage,
        process_env={
            "GITHUB_TOKEN_FILE": str(secret_path),
            "MINDROOM_API_KEY_FILE": str(secret_path),
            "MINDROOM_LOCAL_CLIENT_SECRET_FILE": str(secret_path),
        },
    )
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    worker_runtime_root = worker_root_path(runtime_storage, _TEST_UNSCOPED_WORKER_KEY)

    assert committed_runtime.env_value("GITHUB_TOKEN_FILE") is None
    assert committed_runtime.env_value("MINDROOM_API_KEY_FILE") is None
    assert committed_runtime.env_value("MINDROOM_LOCAL_CLIENT_SECRET_FILE") is None
    assert not (worker_runtime_root / ".runtime" / "file-secrets" / "GITHUB_TOKEN_FILE").exists()
    assert not (worker_runtime_root / ".runtime" / "file-secrets" / "MINDROOM_API_KEY_FILE").exists()
    assert not (worker_runtime_root / ".runtime" / "file-secrets" / "MINDROOM_LOCAL_CLIENT_SECRET_FILE").exists()


def test_docker_backend_excludes_relative_file_backed_secrets_from_worker_payload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated Docker workers must not ambient-copy config-adjacent *_FILE secrets."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    secret_path = config_dir / "secrets" / "openai.key"
    secret_path.parent.mkdir(parents=True, exist_ok=True)
    secret_path.write_text("sk-relative\n", encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    (config_dir / ".env").write_text("OPENAI_API_KEY_FILE=secrets/openai.key\n", encoding="utf-8")
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=runtime_storage)
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    local_secret_copy = (
        worker_root_path(runtime_storage, _TEST_UNSCOPED_WORKER_KEY)
        / ".runtime"
        / "file-secrets"
        / "OPENAI_API_KEY_FILE"
        / "openai.key"
    )

    assert committed_runtime.env_value("OPENAI_API_KEY_FILE") is None
    assert not local_secret_copy.exists()


def test_docker_backend_excludes_relative_process_file_backed_secrets_from_worker_payload(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated Docker workers must not ambient-copy process *_FILE secrets."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    secret_path = config_dir / "secrets" / "openai.key"
    secret_path.parent.mkdir(parents=True, exist_ok=True)
    secret_path.write_text("sk-process-relative\n", encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=runtime_storage,
        process_env={"OPENAI_API_KEY_FILE": "secrets/openai.key"},
    )
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    local_secret_copy = (
        worker_root_path(runtime_storage, _TEST_UNSCOPED_WORKER_KEY)
        / ".runtime"
        / "file-secrets"
        / "OPENAI_API_KEY_FILE"
        / "openai.key"
    )

    assert committed_runtime.env_value("OPENAI_API_KEY_FILE") is None
    assert not local_secret_copy.exists()


def test_docker_backend_preserves_container_config_path_without_host_projection(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated Docker workers without a host config mount must keep the in-container config path."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=tmp_path / "storage")
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=None,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    fake_client = _FakeDockerClient()
    fake_client.images.by_name[config.image] = _FakeImage("sha256:image-v1")

    monkeypatch.setattr(
        "mindroom.workers.backends.docker._load_docker_client_and_errors",
        lambda *_args, **_kwargs: (
            fake_client,
            SimpleNamespace(
                DockerException=_FakeDockerError,
                NotFound=_FakeNotFoundError,
            ),
        ),
    )
    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _noop_sync_shared_credentials,
    )
    backend = DockerWorkerBackend(
        config=config,
        auth_token=_TEST_AUTH_TOKEN,
        storage_path=tmp_path / "storage",
        runtime_paths=runtime_paths,
    )
    monkeypatch.setattr(
        backend,
        "_wait_for_ready",
        lambda container: f"http://127.0.0.1:{backend._container_host_port(container)}/api/sandbox-runner/execute",
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    assert "MINDROOM_CONFIG_PATH" not in env
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    assert committed_runtime.config_path == Path("/app/config-host/config.yaml")


def test_docker_backend_ignores_symlinked_google_application_credentials_path(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Ambient ADC paths are not shared with Docker workers, so no host path copy is attempted."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    real_credentials_path = tmp_path / "real-adc.json"
    real_credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    symlinked_credentials_path = tmp_path / "adc-link.json"
    symlinked_credentials_path.symlink_to(real_credentials_path)
    (config_dir / ".env").write_text(
        f"GOOGLE_APPLICATION_CREDENTIALS={symlinked_credentials_path}\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=runtime_storage,
        process_env={"GOOGLE_APPLICATION_CREDENTIALS": str(symlinked_credentials_path)},
    )
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    assert committed_runtime.env_value("GOOGLE_APPLICATION_CREDENTIALS") is None


def test_docker_backend_does_not_overwrite_google_application_credentials_destination(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Ambient ADC paths must not create or overwrite worker-owned secret files."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_text = "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n"
    config_path.write_text(config_text, encoding="utf-8")
    runtime_storage = (tmp_path / "runtime-storage").resolve()
    credentials_path = tmp_path / "google-credentials.json"
    credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    victim_path = tmp_path / "victim.txt"
    victim_path.write_text("victim\n", encoding="utf-8")
    destination_path = worker_root_path(runtime_storage, _TEST_UNSCOPED_WORKER_KEY) / ".runtime" / credentials_path.name
    destination_path.parent.mkdir(parents=True, exist_ok=True)
    destination_path.symlink_to(victim_path)
    (config_dir / ".env").write_text(
        f"GOOGLE_APPLICATION_CREDENTIALS={credentials_path}\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=runtime_storage,
        process_env={"GOOGLE_APPLICATION_CREDENTIALS": str(credentials_path)},
    )
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=config_text,
        runtime_paths=runtime_paths,
        storage_path=runtime_storage,
        host_config_path=config_path,
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    run_call = fake_client.containers.run_calls[0]
    env = run_call["environment"]
    assert isinstance(env, dict)
    committed_runtime = deserialize_runtime_paths(json.loads(env["MINDROOM_RUNTIME_PATHS_JSON"]))
    assert committed_runtime.env_value("GOOGLE_APPLICATION_CREDENTIALS") is None
    assert destination_path.is_symlink()
    assert victim_path.read_text(encoding="utf-8") == "victim\n"


def test_docker_backend_redacts_projected_config_secrets_and_support_state(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected worker config should omit live secrets and strip unrelated runtime state."""
    config_text, _projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert "api_key" not in projected_config["models"]["default"]
    assert "api_key" not in projected_config["voice"]["stt"]
    assert projected_config["teams"] == {}
    assert projected_config["cultures"] == {}
    assert projected_config["authorization"] == {}
    assert projected_config["matrix_room_access"] == {}
    assert projected_config["mindroom_user"] is None


def test_load_docker_client_auto_installs_optional_runtime(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Docker backend startup should ensure the optional Docker SDK before importing it."""
    captured: dict[str, object] = {}
    fake_client = object()
    fake_errors = SimpleNamespace(DockerException=_FakeDockerError, NotFound=_FakeNotFoundError)

    def _ensure(runtime_paths: RuntimePaths | None = None) -> None:
        captured["installed"] = True
        captured["runtime_paths"] = runtime_paths

    def _import_module(name: str) -> object:
        if name == "docker":
            return SimpleNamespace(from_env=lambda: fake_client)
        if name == "docker.errors":
            return fake_errors
        msg = f"Unexpected import: {name}"
        raise AssertionError(msg)

    monkeypatch.setattr("mindroom.workers.backends.docker.ensure_docker_dependencies", _ensure)
    monkeypatch.setattr("mindroom.workers.backends.docker.importlib.import_module", _import_module)

    client, errors = _load_docker_client_and_errors()

    assert captured["installed"] is True
    assert captured["runtime_paths"] is None
    assert client is fake_client
    assert errors is fake_errors


def test_ensure_docker_dependencies_uses_explicit_runtime_paths(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Docker dependency bootstrap should honor the active runtime's config-adjacent .env."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("MINDROOM_NO_AUTO_INSTALL_TOOLS=true\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(config_path=config_path, process_env={})
    captured: dict[str, object] = {}

    def _ensure_optional_deps(dependencies: list[str], extra_name: str, provided_runtime_paths: RuntimePaths) -> None:
        captured["dependencies"] = dependencies
        captured["extra_name"] = extra_name
        captured["runtime_paths"] = provided_runtime_paths

    monkeypatch.setattr("mindroom.workers.backends.docker.ensure_optional_deps", _ensure_optional_deps)

    ensure_docker_dependencies(runtime_paths)

    assert captured["dependencies"] == ["docker"]
    assert captured["extra_name"] == "docker"
    assert captured["runtime_paths"] is runtime_paths
    assert runtime_paths.env_value("MINDROOM_NO_AUTO_INSTALL_TOOLS") == "true"


def test_read_docker_user_defaults_to_current_posix_uid_gid(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Unset Docker worker user should follow the current POSIX runtime user."""
    monkeypatch.delenv("MINDROOM_DOCKER_WORKER_USER", raising=False)
    monkeypatch.setattr("mindroom.workers.backends.docker_config.os.getuid", lambda: 501)
    monkeypatch.setattr("mindroom.workers.backends.docker_config.os.getgid", lambda: 20)

    assert _default_docker_user_for_os("posix") == "501:20"


def test_read_docker_user_defaults_to_image_user_on_windows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Windows should leave the container user unset and use the image default."""
    monkeypatch.delenv("MINDROOM_DOCKER_WORKER_USER", raising=False)

    assert _default_docker_user_for_os("nt") is None


def test_read_docker_user_env_override_wins(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Explicit Docker worker user config should override platform defaults."""
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_USER", "2001:3001")

    assert _read_docker_user() == "2001:3001"


def test_docker_backend_cleanup_stops_idle_workers(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Idle cleanup should stop running containers while retaining worker state."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, idle_timeout_seconds=60.0)

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    metadata_path = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / "metadata" / "worker.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    metadata["last_used_at"] = 0.0
    metadata["status"] = "ready"
    metadata_path.write_text(json.dumps(metadata), encoding="utf-8")

    cleaned = backend.cleanup_idle_workers(now=100.0)

    assert [worker.worker_key for worker in cleaned] == [_TEST_UNSCOPED_WORKER_KEY]
    assert cleaned[0].status == "idle"

    container = next(iter(fake_client.containers.by_name.values()))
    assert container.stopped == 1
    assert container.status == "exited"

    worker_file = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / "workspace" / "note.txt"
    worker_file.parent.mkdir(parents=True, exist_ok=True)
    worker_file.write_text("still here", encoding="utf-8")
    assert worker_file.read_text(encoding="utf-8") == "still here"


def test_docker_backend_records_failure_and_stops_container(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Recording a failure should stop the worker and persist failure metadata."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)

    handle = backend.record_failure(_TEST_UNSCOPED_WORKER_KEY, "boom", now=11.0)

    assert handle.status == "failed"
    assert handle.failure_count == 1
    assert handle.failure_reason == "boom"

    container = next(iter(fake_client.containers.by_name.values()))
    assert container.stopped == 1
    assert container.status == "exited"


def test_docker_worker_ready_failure_surfaces_container_logs(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Startup log diagnostics should be useful without leaking secrets or huge payloads."""
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    container = _FakeContainer(
        name="mindroom-worker-x",
        image="img",
        image_identity="sha256:x",
        host_port=44999,
        environment={},
        labels={},
        user=None,
        status="exited",
    )
    container.logs_output = (
        b"OPENAI_API_KEY=sk-secret-value\nValidationError: agents.code.display_name Field required\n" + (b"x" * 5000)
    )

    # Call the real implementation; _backend patches the instance attribute.
    with pytest.raises(WorkerBackendError) as exc_info:
        DockerWorkerBackend._wait_for_ready(backend, container)
    message = str(exc_info.value)
    assert "display_name" in message
    assert "sk-secret-value" not in message
    assert "OPENAI_API_KEY=***redacted***" in message
    assert "... [truncated]" in message
    assert len(message) < 4300


def test_docker_worker_health_accepts_matching_protocol() -> None:
    """A worker using the host protocol should pass the compatibility handshake."""
    response = httpx.Response(
        200,
        json={
            "status": "ok",
            "mindroom_version": "2026.8.1",
            "worker_protocol": WORKER_PROTOCOL_VERSION,
        },
    )

    assert _worker_health_compatibility_error(response, image="mindroom:2026.8.1") is None


@pytest.mark.parametrize(
    "payload",
    [
        {"status": "ok"},
        {"status": "ok", "mindroom_version": "2026.7.1", "worker_protocol": 0},
        {"status": "ok", "mindroom_version": "2026.8.1", "worker_protocol": True},
        {"status": "ok", "mindroom_version": "2026.8.1", "worker_protocol": 1.0},
    ],
)
def test_docker_worker_health_rejects_incompatible_protocol(payload: dict[str, object]) -> None:
    """Missing and stale worker protocols should fail with image guidance."""
    response = httpx.Response(200, json=payload)

    error = _worker_health_compatibility_error(response, image="mindroom:stale")

    assert error is not None
    assert "mindroom:stale" in error
    assert f"expected worker protocol {WORKER_PROTOCOL_VERSION}" in error
    assert "Use a worker image built for this MindRoom release" in error


@pytest.mark.parametrize(
    "response",
    [
        pytest.param(httpx.Response(200, text="ok"), id="invalid-json"),
        pytest.param(httpx.Response(200, json=[]), id="non-object-json"),
    ],
)
def test_docker_worker_health_rejects_malformed_payload(response: httpx.Response) -> None:
    """Malformed worker health responses should be treated as incompatible."""
    error = _worker_health_compatibility_error(response, image="mindroom:malformed")

    assert error is not None
    assert "got missing" in error
    assert "worker MindRoom version unknown" in error


def test_docker_worker_readiness_rejects_incompatible_image(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A successful stale health endpoint must not make the worker ready."""
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    container = _FakeContainer(
        name="mindroom-worker-stale",
        image="mindroom:stale",
        image_identity="sha256:stale",
        host_port=44999,
        environment={},
        labels={},
        user=None,
    )

    class _IncompatibleHealthClient:
        def __init__(self, *, timeout: float) -> None:
            assert timeout > 0

        def __enter__(self) -> _IncompatibleHealthClient:
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def get(self, _url: str) -> httpx.Response:
            return httpx.Response(200, json={"status": "ok"})

    monkeypatch.setattr("mindroom.workers.backends.docker.httpx.Client", _IncompatibleHealthClient)

    with pytest.raises(WorkerBackendError, match="expected worker protocol"):
        DockerWorkerBackend._wait_for_ready(backend, container)


def test_docker_worker_stale_image_pulls_and_relaunches(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A stale-image protocol mismatch pulls the configured image and relaunches once."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    fake_client.images.pull_image_id = "sha256:image-v2"
    _use_real_wait_for_ready(monkeypatch, backend)

    _install_health_by_image(monkeypatch, fake_client)

    progress_events: list[WorkerReadyProgress] = []

    handle = backend.ensure_worker(
        WorkerSpec(_TEST_UNSCOPED_WORKER_KEY),
        now=0.0,
        progress_sink=progress_events.append,
    )

    assert fake_client.images.pulls == [backend.config.image]
    assert handle.status == "ready"
    assert handle.startup_count == 2
    assert handle.last_started_at == 0.0
    assert [event.phase for event in progress_events] == ["cold_start", "ready"]
    created = fake_client.containers.created_containers
    assert [item.attrs["Image"] for item in created] == ["sha256:image-v1", "sha256:image-v2"]
    assert created[0].removed == 1


def test_docker_worker_stale_image_recovery_updates_reused_worker_lifecycle(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Replacing a previously ready stale worker records a new startup attempt."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    first_handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    assert first_handle.startup_count == 1

    fake_client.images.pull_image_id = "sha256:image-v2"
    _use_real_wait_for_ready(monkeypatch, backend)
    _install_health_by_image(monkeypatch, fake_client)
    progress_events: list[WorkerReadyProgress] = []

    recovered_handle = backend.ensure_worker(
        WorkerSpec(_TEST_UNSCOPED_WORKER_KEY),
        now=20.0,
        progress_sink=progress_events.append,
    )

    assert recovered_handle.startup_count == 2
    assert recovered_handle.last_started_at == 20.0
    assert [event.phase for event in progress_events] == ["cold_start", "ready"]
    created = fake_client.containers.created_containers
    assert [item.attrs["Image"] for item in created] == ["sha256:image-v1", "sha256:image-v2"]
    assert created[0].removed == 1


def test_docker_worker_stale_image_failed_pull_keeps_compatibility_error(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """When the pull fails, the original compatibility guidance still surfaces."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    _use_real_wait_for_ready(monkeypatch, backend)
    _install_stale_health(monkeypatch)

    with pytest.raises(WorkerBackendError, match="expected worker protocol") as exc_info:
        backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=0.0)

    assert str(exc_info.value).endswith(
        f"Pulling '{backend.config.image}' failed: pull failed: {backend.config.image}",
    )
    assert fake_client.images.pulls == [backend.config.image]


def test_docker_worker_stale_replacement_is_not_retried(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A replacement with the same protocol mismatch fails after one pull."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    fake_client.images.pull_image_id = "sha256:image-v2"
    _use_real_wait_for_ready(monkeypatch, backend)
    _install_stale_health(monkeypatch)

    with pytest.raises(WorkerBackendError, match="expected worker protocol"):
        backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=0.0)

    assert fake_client.images.pulls == [backend.config.image]
    assert len(fake_client.containers.created_containers) == 2


def test_docker_worker_failed_stale_replacement_retains_launch_identity_for_retry(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A failed replacement should be restarted without recreating the pulled image container."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    fake_client.images.pull_image_id = "sha256:image-v2"
    readiness_calls = 0

    def _wait_for_ready(container: _FakeContainer) -> str:
        nonlocal readiness_calls
        readiness_calls += 1
        if container.attrs["Image"] == "sha256:image-v1":
            message = "stale worker image"
            raise _WorkerImageIncompatibleError(message)
        if readiness_calls == 2:
            message = "replacement failed"
            raise WorkerBackendError(message)
        host_port = backend._container_host_port(container)
        return f"http://127.0.0.1:{host_port}/api/sandbox-runner/execute"

    monkeypatch.setattr(backend, "_wait_for_ready", _wait_for_ready)

    with pytest.raises(WorkerBackendError, match="replacement failed"):
        backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=0.0)

    replacement = fake_client.containers.created_containers[-1]
    replacement_hash = replacement.attrs["Config"]["Labels"]["mindroom.ai/launch-config-hash"]
    failed_metadata = backend._load_metadata(backend._state_paths(_TEST_UNSCOPED_WORKER_KEY))
    assert failed_metadata is not None
    assert failed_metadata.launch_config_hash == replacement_hash

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=1.0)

    assert fake_client.containers.by_name[replacement.name] is replacement
    assert replacement.started == 1
    assert replacement.removed == 0
    assert len(fake_client.containers.created_containers) == 2


def test_docker_worker_concurrent_tag_update_keeps_request_local_launch_identity(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Concurrent ensures should converge on the pulled image without sharing launch identity."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)
    recovering_key = _TEST_UNSCOPED_WORKER_KEY
    stale_reader_key = "v1:default:unscoped:research"
    fake_client.images.pull_image_id = "sha256:image-v2"
    race = _ConcurrentTagUpdate(
        backend,
        recovering_key=recovering_key,
        stale_reader_key=stale_reader_key,
    )
    monkeypatch.setattr(backend, "_resolve_launch_config", race.resolve_launch_config)
    monkeypatch.setattr(backend, "_wait_for_ready", race.wait_for_ready)

    with ThreadPoolExecutor(max_workers=2) as executor:
        stale_reader = executor.submit(race.ensure_worker, stale_reader_key)
        assert race.stale_launch_resolved.wait(timeout=5.0)
        recovering_worker = executor.submit(race.ensure_worker, recovering_key)
        recovering_worker.result(timeout=5.0)
        stale_reader.result(timeout=5.0)

    replacement = fake_client.containers.by_name[race.recovering_container_name]
    replacement_labels = replacement.attrs["Config"]["Labels"]
    stale_reader_container = fake_client.containers.by_name[race.stale_reader_container_name]
    stale_reader_labels = stale_reader_container.attrs["Config"]["Labels"]
    recovering_metadata = backend._load_metadata(backend._state_paths(recovering_key))
    stale_reader_metadata = backend._load_metadata(backend._state_paths(stale_reader_key))
    assert recovering_metadata is not None
    assert stale_reader_metadata is not None
    assert replacement.attrs["Image"] == "sha256:image-v2"
    assert recovering_metadata.launch_config_hash == replacement_labels["mindroom.ai/launch-config-hash"]
    assert stale_reader_container.attrs["Image"] == "sha256:image-v2"
    assert stale_reader_metadata.launch_config_hash == stale_reader_labels["mindroom.ai/launch-config-hash"]

    created_count = len(fake_client.containers.created_containers)
    backend.ensure_worker(WorkerSpec(recovering_key), now=1.0)
    backend.ensure_worker(WorkerSpec(stale_reader_key), now=1.0)
    assert len(fake_client.containers.created_containers) == created_count


def test_docker_backend_cleanup_reaps_abandoned_failed_container(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Idle cleanup should reap the exited container of a failed, idle-timed-out worker."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, idle_timeout_seconds=60.0)
    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=0.0)
    backend.record_failure(_TEST_UNSCOPED_WORKER_KEY, "boom", now=1.0)
    container = next(iter(fake_client.containers.by_name.values()))

    # A recently failed worker keeps its exited container for a quick restart.
    backend.cleanup_idle_workers(now=10.0)
    assert container.removed == 0

    # Once past the idle timeout it is abandoned, so the container is reaped.
    backend.cleanup_idle_workers(now=1.0 + 61.0)
    assert container.removed == 1


def test_docker_backend_restart_clears_stale_failure_reason(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Workers that leave the failed state should not keep advertising the old failure message."""
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    backend.record_failure(_TEST_UNSCOPED_WORKER_KEY, "boom", now=11.0)

    restarted = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=12.0)
    assert restarted.status == "ready"
    assert restarted.failure_reason is None


def test_docker_worker_config_rejects_reserved_extra_env_names_from_env(tmp_path: Path) -> None:
    """Docker worker env JSON should not override backend-owned control variables."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")

    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_DOCKER_WORKER_ENV_JSON": json.dumps(
                {
                    "MINDROOM_RUNTIME_PATHS_JSON": "override",
                    "MINDROOM_STORAGE_PATH": str((tmp_path / "escape").resolve()),
                },
            ),
        },
    )

    with pytest.raises(
        WorkerBackendError,
        match="MINDROOM_RUNTIME_PATHS_JSON, MINDROOM_STORAGE_PATH",
    ):
        _DockerWorkerBackendConfig.from_runtime(runtime_paths)


def test_build_dedicated_worker_runtime_paths_rejects_reserved_extra_env_names(tmp_path: Path) -> None:
    """Dedicated worker runtime payloads should reject reserved extra env names."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(config_path=config_path, storage_path=tmp_path)

    with pytest.raises(WorkerBackendError, match="MINDROOM_STORAGE_PATH"):
        build_dedicated_worker_runtime_paths(
            runtime_paths=runtime_paths,
            backend_name="Docker",
            worker_key=_TEST_UNSCOPED_WORKER_KEY,
            config_path=Path("/app/config-host/config.yaml"),
            dedicated_root=Path("/app/worker"),
            worker_port=8766,
            shared_storage_root="/app/shared-storage",
            extra_env={"MINDROOM_STORAGE_PATH": str((tmp_path / "escape").resolve())},
        )


def test_build_dedicated_worker_runtime_paths_rejects_secret_path_override_extra_env(tmp_path: Path) -> None:
    """Dedicated worker extra env must not override rewritten secret-path handoff."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("agents: {}\n", encoding="utf-8")
    secret_path = tmp_path / "secrets" / "openai.key"
    secret_path.parent.mkdir(parents=True, exist_ok=True)
    secret_path.write_text("sk-test\n", encoding="utf-8")
    credentials_path = tmp_path / "adc.json"
    credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "OPENAI_API_KEY_FILE": str(secret_path),
            "GOOGLE_APPLICATION_CREDENTIALS": str(credentials_path),
        },
    )

    with pytest.raises(
        WorkerBackendError,
        match="GOOGLE_APPLICATION_CREDENTIALS, OPENAI_API_KEY_FILE",
    ):
        build_dedicated_worker_runtime_paths(
            runtime_paths=runtime_paths,
            backend_name="Docker",
            worker_key=_TEST_UNSCOPED_WORKER_KEY,
            config_path=Path("/app/config-host/config.yaml"),
            dedicated_root=Path("/app/worker"),
            worker_port=8766,
            shared_storage_root="/app/shared-storage",
            extra_env={
                "GOOGLE_APPLICATION_CREDENTIALS": "/override.json",
                "OPENAI_API_KEY_FILE": "/override.key",
            },
        )


def test_docker_projected_context_files_load_in_worker_runtime(tmp_path: Path) -> None:
    """Projected Docker context files should still load through the worker runtime."""
    config_text, _projected_paths = _projected_config_fixture(tmp_path)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(config_text, encoding="utf-8")
    runtime_paths = resolve_runtime_paths(config_path=config_path, storage_path=tmp_path)
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=config_path,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    manager = DockerProjectionManager(
        config=config,
        projected_configs_root=tmp_path / _PROJECTED_CONFIGS_DIRNAME,
        runtime_paths=runtime_paths,
    )
    worker_paths = local_worker_state_paths_for_root(worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY))
    projection = manager._projected_config(worker_paths, worker_key=_TEST_UNSCOPED_WORKER_KEY, materialize=True)
    projected_config = yaml.safe_load(projection.projected_yaml)
    projected_context_file = projected_config["agents"]["code"]["context_files"][0]
    worker_runtime = build_dedicated_worker_runtime_paths(
        runtime_paths=runtime_paths,
        backend_name="Docker",
        worker_key=_TEST_UNSCOPED_WORKER_KEY,
        config_path=projection.root / "config.yaml",
        dedicated_root=worker_paths.root,
        worker_port=8766,
        shared_storage_root=str(worker_paths.root),
        extra_env={},
    )

    loaded = _load_context_files(
        [projected_context_file],
        worker_runtime,
        agent_name="code",
        storage_path=worker_runtime.storage_root,
    )

    assert len(loaded) == 1
    assert loaded[0].body == "# Context"
    # The title is the worker-visible path, so the rendered prompt names a file
    # the worker runtime can actually open.
    title_path = Path(loaded[0].title)
    assert title_path.is_absolute()
    assert title_path.name == "00-context.md"
    assert title_path.read_text(encoding="utf-8").strip() == "# Context"


@pytest.mark.parametrize("worker_scope", ["shared", "unscoped"])
def test_docker_projection_accepts_normalized_agent_names_in_worker_keys(
    tmp_path: Path,
    worker_scope: str,
) -> None:
    """Projection should match configured agents by the same normalized name used in worker keys."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        (
            """
agents:
  My Agent:
    display_name: Test
    role: Test
    model: default
"""
            + ("    worker_scope: shared\n" if worker_scope == "shared" else "")
            + """
models:
  default:
    provider: openai
    id: test-model
router:
  model: default
"""
        ).lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(config_path=config_path, storage_path=tmp_path)
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=config_path,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    manager = DockerProjectionManager(
        config=config,
        projected_configs_root=tmp_path / _PROJECTED_CONFIGS_DIRNAME,
        runtime_paths=runtime_paths,
    )
    worker_paths = local_worker_state_paths_for_root(tmp_path / "workers" / f"normalized-{worker_scope}")
    if worker_scope == "shared":
        worker_key = resolve_worker_key(
            "shared",
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="My Agent",
                requester_id=None,
                room_id="!room:example.org",
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
                tenant_id="default",
            ),
            agent_name="My Agent",
        )
    else:
        worker_key = resolve_unscoped_worker_key("My Agent")

    projection = manager._projected_config(worker_paths, worker_key=worker_key, materialize=False)
    projected_config = yaml.safe_load(projection.projected_yaml)

    assert list(projected_config["agents"]) == ["My Agent"]


def test_docker_projection_keeps_no_references_to_stripped_agents(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected worker configs must stay loadable after projection strips other agents.

    References to stripped agents fail Config validation inside the worker
    ("calls.agents references unknown agent(s): ..."), crash-looping every
    dedicated worker start.
    """
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
agents:
  alpha:
    display_name: Alpha
    role: Test
    model: default
    worker_scope: shared
    delegate_to: [beta]
  beta:
    display_name: Beta
    role: Test
    model: default
    worker_scope: shared
calls:
  enabled: true
  profiles:
    realtime-profile:
      backend: realtime
      model: test-realtime-model
      credentials_service: openai
      voice: test-voice
  agents:
    alpha: realtime-profile
    beta: realtime-profile
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )

    backend.ensure_worker(WorkerSpec("v1:default:shared:alpha"), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert list(projected_config["agents"]) == ["alpha"]
    assert projected_config["agents"]["alpha"]["delegate_to"] == []
    assert projected_config["calls"] == {}

    projected_runtime_paths = resolve_runtime_paths(
        config_path=projection_root / "config.yaml",
        storage_path=tmp_path / "projected-storage",
    )

    assert set(load_config(projected_runtime_paths).agents) == {"alpha"}


def test_docker_backend_recreates_container_when_launch_config_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Token and launch-config changes should force worker recreation."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    first_handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    first_container = fake_client.containers.by_name[first_handle.worker_id]

    updated_config = replace(
        backend.config,
        user="2000:2000",
        extra_env={"EXTRA_ENV": "updated"},
    )
    updated_backend = DockerWorkerBackend(config=updated_config, auth_token=_ROTATED_AUTH_TOKEN, storage_path=tmp_path)
    monkeypatch.setattr(
        updated_backend,
        "_wait_for_ready",
        lambda container: (
            f"http://127.0.0.1:{updated_backend._container_host_port(container)}/api/sandbox-runner/execute"
        ),
    )

    updated_backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    second_container = fake_client.containers.by_name[first_handle.worker_id]
    assert second_container is not first_container
    assert first_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2

    second_run_call = fake_client.containers.run_calls[-1]
    second_env = second_run_call["environment"]
    assert isinstance(second_env, dict)
    assert second_env["MINDROOM_SANDBOX_PROXY_TOKEN"] == _ROTATED_AUTH_TOKEN
    assert second_env["EXTRA_ENV"] == "updated"
    assert second_run_call["user"] == "2000:2000"


def test_docker_backend_recreates_container_when_name_prefix_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing the configured name prefix should recreate the worker with a new identity."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    first_handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    first_container = fake_client.containers.by_name[first_handle.worker_id]

    updated_config = replace(backend.config, name_prefix="other-prefix")
    updated_backend = DockerWorkerBackend(config=updated_config, auth_token=_TEST_AUTH_TOKEN, storage_path=tmp_path)
    monkeypatch.setattr(
        updated_backend,
        "_wait_for_ready",
        lambda container: (
            f"http://127.0.0.1:{updated_backend._container_host_port(container)}/api/sandbox-runner/execute"
        ),
    )

    second_handle = updated_backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    assert second_handle.worker_id != first_handle.worker_id
    assert first_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2
    assert fake_client.containers.run_calls[0]["name"] != fake_client.containers.run_calls[1]["name"]


def test_docker_backend_uses_distinct_container_names_for_different_storage_roots(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Separate runtimes should not share Docker worker identities."""
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=None,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    fake_client = _FakeDockerClient()

    monkeypatch.setattr(
        "mindroom.workers.backends.docker._load_docker_client_and_errors",
        lambda *_args, **_kwargs: (
            fake_client,
            SimpleNamespace(
                DockerException=_FakeDockerError,
                NotFound=_FakeNotFoundError,
            ),
        ),
    )
    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _noop_sync_shared_credentials,
    )

    first_storage_root = tmp_path / "runtime-a"
    second_storage_root = tmp_path / "runtime-b"
    first_backend = DockerWorkerBackend(config=config, auth_token=_TEST_AUTH_TOKEN, storage_path=first_storage_root)
    monkeypatch.setattr(
        first_backend,
        "_wait_for_ready",
        lambda container: (
            f"http://127.0.0.1:{first_backend._container_host_port(container)}/api/sandbox-runner/execute"
        ),
    )

    second_backend = DockerWorkerBackend(config=config, auth_token=_TEST_AUTH_TOKEN, storage_path=second_storage_root)
    monkeypatch.setattr(
        second_backend,
        "_wait_for_ready",
        lambda container: (
            f"http://127.0.0.1:{second_backend._container_host_port(container)}/api/sandbox-runner/execute"
        ),
    )

    first_handle = first_backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    second_handle = second_backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    assert first_handle.worker_id != second_handle.worker_id
    assert len(fake_client.containers.run_calls) == 2
    assert fake_client.containers.run_calls[0]["name"] != fake_client.containers.run_calls[1]["name"]


def test_docker_backend_shutdown_removes_running_containers_and_keeps_idle_metadata(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Backend shutdown should remove managed containers before the manager is discarded."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    container = fake_client.containers.by_name[handle.worker_id]

    backend.shutdown()

    assert container.removed == 1
    idle_handle = backend.list_workers(now=20.0)[0]
    assert idle_handle.status == "idle"
    assert idle_handle.failure_reason is None


def test_docker_backend_recreates_container_when_storage_mount_does_not_match(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Worker reuse should fail closed when an existing container points at the wrong state root."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    existing_container.attrs["Mounts"][0]["Source"] = str(tmp_path / "wrong-root")

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2


def test_docker_backend_recreates_container_when_extra_stale_mount_is_present(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Worker reuse should fail closed when a stale container keeps extra bind mounts."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = "v1:tenant-123:user_agent:@alice:example.org:alpha"

    handle = backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    existing_container.attrs["Mounts"].append(
        {
            "Type": "bind",
            "Source": str((tmp_path / "agents" / "beta").resolve()),
            "Destination": "/app/worker/agents/beta",
            "Mode": "rw",
            "RW": True,
        },
    )

    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2


def test_docker_backend_ignores_container_runtime_non_bind_mounts(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Anonymous volumes and tmpfs mounts should not force otherwise compatible workers to restart."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    existing_container.attrs["Mounts"].append(
        {
            "Type": "volume",
            "Name": "anonymous-volume",
            "Destination": "/var/lib/mindroom/runtime-volume",
            "Mode": "",
            "RW": True,
        },
    )

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    assert fake_client.containers.by_name[handle.worker_id] is existing_container
    assert existing_container.removed == 0
    assert len(fake_client.containers.run_calls) == 1


def test_docker_backend_recreates_container_when_host_config_contents_change(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Worker reuse should fail closed when the mounted config file changes contents."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]

    (tmp_path / "config.yaml").write_text("agents:\n  code:\n    tools: [shell]\n", encoding="utf-8")

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2


def test_docker_backend_recreates_container_when_projected_file_asset_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing a projected single-file asset should rotate the worker."""
    config_text, projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    first_volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(first_volumes, dict)
    first_projection_root = _projection_root(first_volumes)

    updated_context_file = projected_paths["context_file"].with_suffix(".updated.md")
    updated_context_file.write_text("# Updated Context\n", encoding="utf-8")
    updated_context_file.replace(projected_paths["context_file"])

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2

    second_volumes = fake_client.containers.run_calls[-1]["volumes"]
    assert isinstance(second_volumes, dict)
    second_projection_root = _projection_root(second_volumes)
    assert second_projection_root != first_projection_root
    projected_context_path = (
        second_projection_root / ".mindroom-worker-assets" / "agents" / "code" / "context_files" / "00-context.md"
    )
    assert projected_context_path.read_text(encoding="utf-8") == "# Updated Context\n"


def test_docker_backend_recreates_container_when_projected_directory_asset_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing a projected directory asset should rotate the worker."""
    config_text, projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    first_volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(first_volumes, dict)
    first_projection_root = _projection_root(first_volumes)

    replacement_knowledge_root = projected_paths["knowledge_root"].with_name("knowledge_docs.updated")
    replacement_knowledge_root.mkdir()
    (replacement_knowledge_root / "guide.md").write_text("# Guide v2\n", encoding="utf-8")
    archived_knowledge_root = projected_paths["knowledge_root"].with_name("knowledge_docs.previous")
    projected_paths["knowledge_root"].replace(archived_knowledge_root)
    replacement_knowledge_root.replace(projected_paths["knowledge_root"])

    removal_checks: list[bool] = []
    original_remove_container = backend._remove_container

    def _assert_old_projection_survives_until_removal(container: object) -> None:
        assert container is existing_container
        removal_checks.append(first_projection_root.exists())
        original_remove_container(container)

    monkeypatch.setattr(backend, "_remove_container", _assert_old_projection_survives_until_removal)

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert removal_checks == [True]
    assert len(fake_client.containers.run_calls) == 2

    second_volumes = fake_client.containers.run_calls[-1]["volumes"]
    assert isinstance(second_volumes, dict)
    second_projection_root = _projection_root(second_volumes)
    assert second_projection_root != first_projection_root
    assert not first_projection_root.exists()
    projected_knowledge_path = (
        second_projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "docs" / "guide.md"
    )
    assert projected_knowledge_path.read_text(encoding="utf-8") == "# Guide v2\n"


def test_docker_backend_projects_only_agent_specific_assets_for_shared_worker(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Agent-scoped workers should not snapshot unrelated agents' projected assets."""
    config_text, projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = "v1:default:shared:alpha"

    backend.ensure_worker(WorkerSpec(worker_key), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = (projection_root / "config.yaml").read_text(encoding="utf-8")
    projected_config_data = yaml.safe_load(projected_config)

    assert "- ./.mindroom-worker-assets/agents/alpha/context_files/00-alpha.md" in projected_config
    assert ".mindroom-worker-assets/agents/beta/context_files/00-beta.md" not in projected_config
    assert "path: ./.mindroom-worker-assets/knowledge_bases/a" in projected_config
    assert "path: ./.mindroom-worker-assets/knowledge_bases/b" not in projected_config
    assert set(projected_config_data["agents"]) == {"alpha"}
    assert set(projected_config_data["knowledge_bases"]) == {"a"}

    projected_alpha_context = (
        projection_root / ".mindroom-worker-assets" / "agents" / "alpha" / "context_files" / "00-alpha.md"
    )
    assert projected_alpha_context.read_text(encoding="utf-8") == "# Alpha\n"
    assert not (
        projection_root / ".mindroom-worker-assets" / "agents" / "beta" / "context_files" / "00-beta.md"
    ).exists()

    projected_alpha_knowledge = projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "a" / "a.txt"
    assert projected_alpha_knowledge.read_text(encoding="utf-8") == "alpha knowledge\n"
    assert not (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "b" / "b.txt").exists()

    assert projected_paths["alpha_context"].resolve() not in projection_root.parents
    assert projected_paths["beta_context"].resolve() not in projection_root.parents
    assert projected_paths["alpha_knowledge_root"].resolve() not in projection_root.parents
    assert projected_paths["beta_knowledge_root"].resolve() not in projection_root.parents


def test_docker_backend_projects_context_files_from_canonical_agent_workspace(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Shared workers should project context files from the agent workspace, not the config dir."""
    (tmp_path / "README.md").write_text("CONFIG ROOT FILE\n", encoding="utf-8")
    workspace_readme = tmp_path / "agents" / "alpha" / "workspace" / "README.md"
    workspace_readme.parent.mkdir(parents=True)
    workspace_readme.write_text("AGENT WORKSPACE FILE\n", encoding="utf-8")
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: shared
    context_files:
      - README.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )

    backend.ensure_worker(WorkerSpec("v1:default:shared:alpha"), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_readme = (
        projection_root / ".mindroom-worker-assets" / "agents" / "alpha" / "context_files" / "00-README.md"
    )

    assert projected_readme.read_text(encoding="utf-8") == "AGENT WORKSPACE FILE\n"


def test_docker_backend_projects_only_private_user_agent_assets_for_private_agent(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Private user-agent workers should not snapshot unrelated agents or knowledge bases."""
    config_text, projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )
    assert worker_key is not None

    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config_data = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert set(projected_config_data["agents"]) == {"alpha"}
    assert set(projected_config_data["knowledge_bases"]) == {"a"}
    assert (
        projection_root / ".mindroom-worker-assets" / "agents" / "alpha" / "context_files" / "00-alpha.md"
    ).read_text(encoding="utf-8") == "# Alpha\n"
    assert not (
        projection_root / ".mindroom-worker-assets" / "agents" / "beta" / "context_files" / "00-beta.md"
    ).exists()
    assert (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "a" / "a.txt").read_text(
        encoding="utf-8",
    ) == "alpha knowledge\n"
    assert not (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "b" / "b.txt").exists()

    assert projected_paths["alpha_context"].resolve() not in projection_root.parents
    assert projected_paths["beta_context"].resolve() not in projection_root.parents
    assert projected_paths["alpha_knowledge_root"].resolve() not in projection_root.parents
    assert projected_paths["beta_knowledge_root"].resolve() not in projection_root.parents


def test_docker_backend_projects_private_template_dirs_for_private_agents(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected configs should keep private template scaffolds usable inside Docker workers."""
    template_dir = tmp_path / "template"
    template_dir.mkdir()
    (template_dir / "README.md").write_text("template scaffold\n", encoding="utf-8")
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
agents:
  alpha:
    display_name: Alpha
    role: Test
    model: default
    private:
      per: user_agent
      template_dir: ./template
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )
    assert worker_key is not None

    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert projected_config["agents"]["alpha"]["private"]["template_dir"] == (
        "./.mindroom-worker-assets/agents/alpha/private/template_dir"
    )
    assert (
        projection_root / ".mindroom-worker-assets" / "agents" / "alpha" / "private" / "template_dir" / "README.md"
    ).read_text(encoding="utf-8") == "template scaffold\n"

    projected_runtime_paths = resolve_runtime_paths(
        config_path=projection_root / "config.yaml",
        storage_path=tmp_path / "projected-storage",
    )
    projected_runtime_config = load_config(projected_runtime_paths)
    workspace = resolve_agent_workspace_from_state_path(
        "alpha",
        projected_runtime_config,
        runtime_paths=projected_runtime_paths,
        state_storage_path=tmp_path / "private-root",
        use_state_storage_path=True,
        create=True,
    )

    assert workspace is not None
    assert (workspace.root / "README.md").read_text(encoding="utf-8") == "template scaffold\n"


def test_docker_backend_shared_worker_mounts_canonical_agent_root(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Shared workers should mount the canonical shared agent root into the container."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "shared",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )

    backend.ensure_worker(WorkerSpec(worker_key), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    expected_agent_root = (tmp_path / "agents" / "alpha").resolve()
    assert volumes[str(expected_agent_root)] == {
        "bind": "/app/worker/agents/alpha",
        "mode": "rw",
    }
    env = fake_client.containers.run_calls[0]["environment"]
    assert isinstance(env, dict)
    assert env["HOME"] == "/app/worker/agents/alpha/workspace"


def test_docker_backend_user_agent_requires_explicit_private_visibility(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """User-agent workers should fail closed without an explicit private-agent visibility set."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )

    with pytest.raises(WorkerBackendError, match="user_agent workers require explicit private-agent visibility"):
        backend.ensure_worker(WorkerSpec(worker_key), now=10.0)


def test_docker_backend_user_agent_home_is_neutral_without_visibility(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """User-agent HOME should come from worker-key scope, not visibility metadata."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, _fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )

    assert worker_key is not None
    assert backend._container_env(worker_key)["HOME"] == "/app/worker"


def test_docker_backend_rejects_unknown_worker_keys_for_scoped_mounts(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Malformed worker keys must not fall back to mounting the whole storage root."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="Unsupported worker key"):
        backend.ensure_worker(WorkerSpec("not-a-valid-worker-key"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_stale_scoped_worker_keys_for_projection(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Scoped worker keys must fail closed when they no longer resolve to any configured agent."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(WorkerSpec("v1:default:shared:missing"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_scoped_worker_keys_that_no_longer_match_agent_policy(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A stale shared worker key must not broaden projection after an agent switches to user_agent isolation."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(WorkerSpec("v1:default:shared:alpha"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_unknown_unscoped_worker_keys_for_projection(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Unknown unscoped worker keys must not broaden the projected config."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(WorkerSpec("v1:default:unscoped:missing"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_stale_unscoped_worker_keys_for_projection(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A stale unscoped worker key must fail closed after an agent changes isolation scope."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(WorkerSpec("v1:default:unscoped:alpha"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_stale_user_agent_worker_keys_for_projection(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A stale user-agent worker key must fail closed instead of broadening projection."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(
            WorkerSpec(
                "v1:tenant-123:user_agent:@alice:example.org:missing",
                private_agent_names=frozenset({"missing"}),
            ),
            now=10.0,
        )
    assert fake_client.containers.run_calls == []


def test_docker_backend_rejects_user_agent_worker_keys_when_agent_is_shared(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A user-agent worker key must fail closed when the agent still exists but is not user_agent scoped."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(
            WorkerSpec(
                "v1:default:user_agent:@alice:example.org:alpha",
                private_agent_names=frozenset({"alpha"}),
            ),
            now=10.0,
        )
    assert fake_client.containers.run_calls == []


def test_docker_backend_projects_only_user_scoped_assets_for_requester_worker(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Requester-scoped Docker workers should only snapshot agents that resolve to worker_scope=user."""
    for name, contents in {
        "alpha": "alpha knowledge\n",
        "beta": "beta knowledge\n",
        "gamma": "gamma knowledge\n",
        "delta": "delta knowledge\n",
    }.items():
        knowledge_root = tmp_path / f"knowledge_{name}"
        knowledge_root.mkdir()
        (knowledge_root / f"{name}.txt").write_text(contents, encoding="utf-8")
        context_path = tmp_path / "agents" / name / "workspace" / f"{name}.md"
        context_path.parent.mkdir(parents=True, exist_ok=True)
        context_path.write_text(f"# {name.title()}\n", encoding="utf-8")

    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
knowledge_bases:
  a:
    path: ./knowledge_alpha
  b:
    path: ./knowledge_beta
  c:
    path: ./knowledge_gamma
  d:
    path: ./knowledge_delta
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: user
    knowledge_bases: [a]
    context_files:
      - alpha.md
  beta:
    display_name: Beta
    role: Beta test
    model: default
    worker_scope: shared
    knowledge_bases: [b]
    context_files:
      - beta.md
  gamma:
    display_name: Gamma
    role: Gamma test
    model: default
    knowledge_bases: [c]
    context_files:
      - gamma.md
  delta:
    display_name: Delta
    role: Delta test
    model: default
    private:
      per: user
    knowledge_bases: [d]
    context_files:
      - delta.md
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
    )

    backend.ensure_worker(WorkerSpec("v1:tenant-123:user:@alice:example.org"), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config_data = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert set(projected_config_data["agents"]) == {"alpha", "delta"}
    assert set(projected_config_data["knowledge_bases"]) == {"a", "d"}
    assert (
        projection_root / ".mindroom-worker-assets" / "agents" / "alpha" / "context_files" / "00-alpha.md"
    ).read_text(encoding="utf-8") == "# Alpha\n"
    assert (
        projection_root / ".mindroom-worker-assets" / "agents" / "delta" / "context_files" / "00-delta.md"
    ).read_text(encoding="utf-8") == "# Delta\n"
    assert not (
        projection_root / ".mindroom-worker-assets" / "agents" / "beta" / "context_files" / "00-beta.md"
    ).exists()
    assert not (
        projection_root / ".mindroom-worker-assets" / "agents" / "gamma" / "context_files" / "00-gamma.md"
    ).exists()
    assert (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "a" / "alpha.txt").read_text(
        encoding="utf-8",
    ) == "alpha knowledge\n"
    assert (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "d" / "delta.txt").read_text(
        encoding="utf-8",
    ) == "delta knowledge\n"
    assert not (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "b" / "beta.txt").exists()
    assert not (projection_root / ".mindroom-worker-assets" / "knowledge_bases" / "c" / "gamma.txt").exists()


def test_docker_backend_rejects_user_worker_keys_without_user_scoped_agents(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Requester-scoped workers must fail closed when no configured agent resolves to worker_scope=user."""
    config_text, _projected_paths = _multi_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    with pytest.raises(WorkerBackendError, match="does not match any configured agent policy"):
        backend.ensure_worker(WorkerSpec("v1:tenant-123:user:@alice:example.org"), now=10.0)
    assert fake_client.containers.run_calls == []


def test_docker_backend_user_agent_mounts_private_root_from_worker_spec(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """User-agent workers should mount their private instance root when explicitly visible."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )

    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    expected_private_root = (tmp_path / "private_instances" / worker_dir_name(worker_key) / "alpha").resolve()
    assert volumes[str(expected_private_root)] == {
        "bind": f"/app/worker/private_instances/{worker_dir_name(worker_key)}/alpha",
        "mode": "rw",
    }
    assert all(spec["bind"] != "/app/worker/agents/alpha" for spec in volumes.values())
    env = fake_client.containers.run_calls[0]["environment"]
    assert isinstance(env, dict)
    assert env["HOME"] == "/app/worker"


def test_docker_backend_rejects_private_user_agent_container_without_target_visibility(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Private user-agent workers must fail closed until the targeted private agent is explicitly visible."""
    config_text, _projected_paths = _private_user_agent_projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="alpha",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
        ),
        agent_name="alpha",
    )

    with pytest.raises(WorkerBackendError, match="missing from explicit private-agent visibility"):
        backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset()), now=10.0)
    assert fake_client.containers.run_calls == []

    handle = backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"alpha"})), now=20.0)
    assert len(fake_client.containers.run_calls) == 1
    second_volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(second_volumes, dict)
    expected_private_root = (tmp_path / "private_instances" / worker_dir_name(worker_key) / "alpha").resolve()
    assert second_volumes[str(expected_private_root)] == {
        "bind": f"/app/worker/private_instances/{worker_dir_name(worker_key)}/alpha",
        "mode": "rw",
    }
    assert handle.status == "ready"


def test_docker_backend_redacts_authorization_headers_in_projected_model_extra_kwargs(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected config should drop auth headers nested under model extra_kwargs."""
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text="""
agents:
  code:
    display_name: Code
    role: Test
    model: default
    worker_scope: shared
models:
  default:
    provider: openai
    id: test-model
    extra_kwargs:
      headers:
        Authorization: Bearer super-secret-token
        X-Trace-Id: keep-me
""".lstrip(),
    )

    backend.ensure_worker(WorkerSpec("v1:default:shared:code"), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_config = yaml.safe_load((projection_root / "config.yaml").read_text(encoding="utf-8"))

    assert "Authorization" not in projected_config["models"]["default"]["extra_kwargs"]["headers"]
    assert projected_config["models"]["default"]["extra_kwargs"]["headers"]["X-Trace-Id"] == "keep-me"


def test_docker_backend_reconciles_missing_container_metadata_without_stale_endpoint(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Missing containers should not keep advertising a stale published endpoint."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, idle_timeout_seconds=60.0)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    container_name = next(iter(fake_client.containers.by_name))
    del fake_client.containers.by_name[container_name]

    handles = backend.list_workers(include_idle=True, now=100.0)

    assert [worker.worker_key for worker in handles] == [_TEST_UNSCOPED_WORKER_KEY]
    assert handles[0].status == "idle"
    assert handles[0].endpoint == "/api/sandbox-runner/execute"
    assert handle.endpoint != handles[0].endpoint

    metadata_path = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / "metadata" / "worker.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["host_port"] is None
    assert metadata["container_id"] is None

    cleaned = backend.cleanup_idle_workers(now=1000.0)

    assert cleaned == []
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["host_port"] is None
    assert metadata["container_id"] is None


def test_docker_projection_hash_is_stable_across_hash_seeds(tmp_path: Path) -> None:
    """Projection hashes should not change across interpreter restarts for the same config."""
    (tmp_path / "kb_a").mkdir()
    (tmp_path / "kb_a" / "a.txt").write_text("a\n", encoding="utf-8")
    (tmp_path / "kb_b").mkdir()
    (tmp_path / "kb_b" / "b.txt").write_text("b\n", encoding="utf-8")
    (tmp_path / "config.yaml").write_text(
        """
knowledge_bases:
  a:
    path: ./kb_a
  b:
    path: ./kb_b
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: shared
    knowledge_bases: [a, b]
models:
  default:
    provider: openai
    id: test-model
""".lstrip(),
        encoding="utf-8",
    )

    first_signature = _projection_signature_for_hash_seed("1", tmp_path)
    second_signature = _projection_signature_for_hash_seed("2", tmp_path)

    assert first_signature == second_signature
    assert first_signature["knowledge_bases"] == ["a", "b"]


def test_docker_projection_hash_changes_when_container_config_filename_changes(tmp_path: Path) -> None:
    """Renaming the container-side config file should materialize a fresh projection snapshot."""
    host_config_path = tmp_path / "config.yaml"
    host_config_path.write_text(
        """
models:
  default:
    provider: openai
    id: test-model
agents: {}
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(config_path=host_config_path, storage_path=tmp_path)
    paths = local_worker_state_paths_for_root(tmp_path / "workers" / "projection-test")
    base_config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=host_config_path,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )

    first_manager = DockerProjectionManager(
        config=base_config,
        projected_configs_root=tmp_path / "projections",
        runtime_paths=runtime_paths,
    )
    first_projection = first_manager._projected_config(paths, materialize=True)

    renamed_manager = DockerProjectionManager(
        config=replace(base_config, config_path="/app/config-host/alt.yaml"),
        projected_configs_root=tmp_path / "projections",
        runtime_paths=runtime_paths,
    )
    renamed_projection = renamed_manager._projected_config(paths, materialize=True)

    assert renamed_projection.root != first_projection.root
    assert not first_projection.root.exists()
    assert (renamed_projection.root / "alt.yaml").is_file()
    assert not (renamed_projection.root / "config.yaml").exists()


def test_docker_projection_hash_changes_when_projected_file_mode_changes(tmp_path: Path) -> None:
    """Changing only a projected file mode should rebuild the snapshot and preserve the new mode."""
    plugin_dir = tmp_path / "plugins" / "demo"
    plugin_dir.mkdir(parents=True)
    plugin_file = plugin_dir / "helper.sh"
    plugin_file.write_text("#!/bin/sh\necho hi\n", encoding="utf-8")
    plugin_file.chmod(0o644)

    host_config_path = tmp_path / "config.yaml"
    host_config_path.write_text(
        """
plugins:
  - ./plugins/demo
models:
  default:
    provider: openai
    id: test-model
agents: {}
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(config_path=host_config_path, storage_path=tmp_path)
    paths = local_worker_state_paths_for_root(tmp_path / "workers" / "projection-test")
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=host_config_path,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    manager = DockerProjectionManager(
        config=config,
        projected_configs_root=tmp_path / "projections",
        runtime_paths=runtime_paths,
    )

    first_projection = manager._projected_config(paths, materialize=True)
    first_projected_file = first_projection.root / ".mindroom-worker-assets" / "plugins" / "00-demo" / "helper.sh"
    assert first_projected_file.stat().st_mode & 0o777 == 0o644

    plugin_file.chmod(0o755)

    second_projection = manager._projected_config(paths, materialize=True)
    second_projected_file = second_projection.root / ".mindroom-worker-assets" / "plugins" / "00-demo" / "helper.sh"

    assert second_projection.root != first_projection.root
    assert not first_projection.root.exists()
    assert second_projected_file.stat().st_mode & 0o777 == 0o755


def test_docker_projection_hash_changes_when_projected_directory_mode_changes(tmp_path: Path) -> None:
    """Changing only a projected directory mode should rebuild the snapshot and preserve the new mode."""
    plugin_dir = tmp_path / "plugins" / "demo"
    plugin_dir.mkdir(parents=True)
    plugin_dir.chmod(0o755)
    (plugin_dir / "helper.sh").write_text("#!/bin/sh\necho hi\n", encoding="utf-8")

    host_config_path = tmp_path / "config.yaml"
    host_config_path.write_text(
        """
plugins:
  - ./plugins/demo
models:
  default:
    provider: openai
    id: test-model
agents: {}
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(config_path=host_config_path, storage_path=tmp_path)
    paths = local_worker_state_paths_for_root(tmp_path / "workers" / "projection-test")
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=host_config_path,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    manager = DockerProjectionManager(
        config=config,
        projected_configs_root=tmp_path / "projections",
        runtime_paths=runtime_paths,
    )

    first_projection = manager._projected_config(paths, materialize=True)
    first_projected_dir = first_projection.root / ".mindroom-worker-assets" / "plugins" / "00-demo"
    assert first_projected_dir.stat().st_mode & 0o777 == 0o755

    plugin_dir.chmod(0o700)

    second_projection = manager._projected_config(paths, materialize=True)
    second_projected_dir = second_projection.root / ".mindroom-worker-assets" / "plugins" / "00-demo"

    assert second_projection.root != first_projection.root
    assert not first_projection.root.exists()
    assert second_projected_dir.stat().st_mode & 0o777 == 0o700


def test_docker_backend_rebuilds_incomplete_projection_snapshot(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Interrupted projection writes should be repaired instead of being reused forever."""
    config_text, _projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    first_volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(first_volumes, dict)
    projection_root = _projection_root(first_volumes)

    (projection_root / ".projection-ready").unlink()
    (projection_root / "config.yaml").write_text("broken: true\n", encoding="utf-8")

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    second_volumes = fake_client.containers.run_calls[-1]["volumes"]
    assert isinstance(second_volumes, dict)
    second_projection_root = _projection_root(second_volumes)
    replacement_container = fake_client.containers.by_name[handle.worker_id]

    assert second_projection_root == projection_root
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert (projection_root / ".projection-ready").read_text(encoding="utf-8") == "ready\n"
    rebuilt_config = (projection_root / "config.yaml").read_text(encoding="utf-8")
    assert "broken: true" not in rebuilt_config
    assert "plugins:" in rebuilt_config


def test_docker_backend_rebuilds_corrupted_ready_projection_snapshot(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A leftover ready marker must not make the backend trust a projection missing required files."""
    config_text, _projected_paths = _projected_config_fixture(tmp_path)
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path, config_text=config_text)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]
    first_volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(first_volumes, dict)
    projection_root = _projection_root(first_volumes)

    (projection_root / "config.yaml").unlink()
    assert (projection_root / ".projection-ready").is_file()

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2
    assert (projection_root / ".projection-ready").read_text(encoding="utf-8") == "ready\n"
    rebuilt_config = (projection_root / "config.yaml").read_text(encoding="utf-8")
    assert "plugins:" in rebuilt_config
    assert "memory:" in rebuilt_config


def test_docker_backend_disambiguates_colliding_projected_knowledge_base_ids(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Projected knowledge-base directories should stay unique when IDs sanitize the same way."""
    backend, fake_client, _sync_calls = _backend(
        monkeypatch,
        tmp_path,
        config_text=_knowledge_base_collision_fixture(tmp_path),
    )

    backend.ensure_worker(WorkerSpec("v1:default:shared:alpha"), now=10.0)

    volumes = fake_client.containers.run_calls[0]["volumes"]
    assert isinstance(volumes, dict)
    projection_root = _projection_root(volumes)
    projected_knowledge_root = projection_root / ".mindroom-worker-assets" / "knowledge_bases"
    projected_dirs = sorted(path.name for path in projected_knowledge_root.iterdir())

    assert "docs-a" in projected_dirs
    assert any(path_name.startswith("docs-a-") for path_name in projected_dirs)
    assert len(projected_dirs) == 2
    assert (projected_knowledge_root / "docs-a" / "a.txt").read_text(encoding="utf-8") == "first\n"
    colliding_dir = next(path_name for path_name in projected_dirs if path_name != "docs-a")
    assert (projected_knowledge_root / colliding_dir / "b.txt").read_text(encoding="utf-8") == "second\n"


def test_docker_backend_reuses_container_after_first_run_pulls_missing_image(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The first auto-pull should not force a second worker recreation on the next ensure."""
    config = _DockerWorkerBackendConfig(
        image="ghcr.io/mindroom-ai/mindroom:latest",
        worker_port=8766,
        storage_mount_path="/app/worker",
        config_path="/app/config-host/config.yaml",
        host_config_path=None,
        idle_timeout_seconds=60.0,
        ready_timeout_seconds=5.0,
        name_prefix="mindroom-worker",
        publish_host="127.0.0.1",
        endpoint_host="127.0.0.1",
        user="1000:1000",
        extra_env={},
        extra_labels={},
    )
    fake_client = _FakeDockerClient(auto_pull_missing_image=True)

    monkeypatch.setattr(
        "mindroom.workers.backends.docker._load_docker_client_and_errors",
        lambda *_args, **_kwargs: (
            fake_client,
            SimpleNamespace(
                DockerException=_FakeDockerError,
                NotFound=_FakeNotFoundError,
            ),
        ),
    )
    monkeypatch.setattr(
        "mindroom.workers.backends.docker.sync_shared_credentials_to_worker",
        _noop_sync_shared_credentials,
    )

    backend = DockerWorkerBackend(config=config, auth_token=_TEST_AUTH_TOKEN, storage_path=tmp_path)
    monkeypatch.setattr(
        backend,
        "_wait_for_ready",
        lambda container: f"http://127.0.0.1:{backend._container_host_port(container)}/api/sandbox-runner/execute",
    )

    first_handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    first_container = fake_client.containers.by_name[first_handle.worker_id]

    second_handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    assert second_handle.worker_id == first_handle.worker_id
    assert fake_client.containers.by_name[second_handle.worker_id] is first_container
    assert len(fake_client.containers.run_calls) == 1
    assert first_container.removed == 0
    metadata_path = worker_root_path(tmp_path, _TEST_UNSCOPED_WORKER_KEY) / "metadata" / "worker.json"
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    assert metadata["startup_count"] == 1
    assert metadata["last_started_at"] == 10.0
    assert metadata["last_used_at"] == 20.0
    assert metadata["status"] == "ready"


def test_docker_backend_recreates_container_when_same_tag_resolves_to_new_image_id(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Rebuilding the same image tag locally should rotate the worker on the next ensure."""
    backend, fake_client, _sync_calls = _backend(monkeypatch, tmp_path)

    handle = backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=10.0)
    existing_container = fake_client.containers.by_name[handle.worker_id]

    fake_client.images.by_name[backend.config.image] = _FakeImage("sha256:image-v2")

    backend.ensure_worker(WorkerSpec(_TEST_UNSCOPED_WORKER_KEY), now=20.0)

    replacement_container = fake_client.containers.by_name[handle.worker_id]
    assert replacement_container is not existing_container
    assert existing_container.removed == 1
    assert len(fake_client.containers.run_calls) == 2
