"""Tests for the Kubernetes worker backend."""

from __future__ import annotations

import base64
import hashlib
import json
import threading
import time
from contextlib import nullcontext
from copy import deepcopy
from pathlib import Path
from types import MethodType, SimpleNamespace
from typing import TYPE_CHECKING, Self
from unittest.mock import patch

import pytest

from mindroom.config.main import load_config
from mindroom.config.yaml_includes import load_yaml_config_source
from mindroom.constants import (
    DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    deserialize_runtime_paths,
    resolve_primary_runtime_paths,
    sandbox_startup_manifest_path,
    startup_manifest_sha256,
)
from mindroom.runtime_env_policy import CREDENTIALS_ENCRYPTION_KEY_ENV
from mindroom.tool_system.worker_routing import (
    ToolExecutionIdentity,
    _private_instance_state_root_path,
    descriptive_worker_id_for_key,
    resolve_unscoped_worker_key,
    resolve_worker_key,
    worker_dir_name,
)
from mindroom.workers.backend import WorkerBackendError
from mindroom.workers.backends import kubernetes as kubernetes_backend_module
from mindroom.workers.backends import kubernetes_resources as kubernetes_resources_module
from mindroom.workers.backends.kubernetes import (
    KubernetesWorkerBackend,
    KubernetesWorkerBackendConfig,
    kubernetes_backend_config_signature,
)
from mindroom.workers.backends.kubernetes_config import KubernetesAgentVaultConfig
from mindroom.workers.backends.kubernetes_resources import (
    _ANNOTATION_CREDENTIALS_ENCRYPTION_KEY_HASH,
    _ANNOTATION_PRIVATE_AGENT_NAMES,
    _ANNOTATION_RUNNER_TOKEN_HASH,
    _ANNOTATION_STARTUP_MANIFEST_HASH,
    _ANNOTATION_TEMPLATE_HASH,
    ANNOTATION_WORKER_KEY,
    worker_auth_token,
)
from mindroom.workers.models import WorkerReadyProgress, WorkerSpec
from mindroom.workers.runtime import (
    primary_worker_backend_available,
    primary_worker_backend_name,
    serialized_kubernetes_worker_config_snapshot,
)

if TYPE_CHECKING:
    from collections.abc import Callable

    from mindroom.constants import RuntimePaths

_TEST_AUTH_TOKEN = "test-token"  # noqa: S105
_TEST_SCOPED_WORKER_KEY_A = "v1:tenant-123:shared:code"
_TEST_SCOPED_WORKER_KEY_B = "v1:tenant-123:shared:research"
_TEST_TOOL_VALIDATION_SNAPSHOT = {
    "calculator": {
        "config_fields": [],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    },
}


class _ControlledMonotonicClock:
    def __init__(self, initial_seconds: float = 0.0) -> None:
        self._now = initial_seconds
        self._condition = threading.Condition()
        self._listeners: list[Callable[[], None]] = []

    def monotonic(self) -> float:
        with self._condition:
            return self._now

    def sleep(self, seconds: float) -> None:
        self.wait_until(self.monotonic() + seconds)

    def wait_until(self, target_seconds: float) -> None:
        with self._condition:
            while self._now < target_seconds:
                self._condition.wait(timeout=0.1)

    def advance_to(self, target_seconds: float) -> None:
        with self._condition:
            assert target_seconds >= self._now
            self._now = target_seconds
            listeners = tuple(self._listeners)
            self._condition.notify_all()
        for listener in listeners:
            listener()

    def add_listener(self, listener: Callable[[], None]) -> None:
        with self._condition:
            self._listeners.append(listener)


class _ControlledCondition:
    def __init__(self, clock: _ControlledMonotonicClock) -> None:
        self._clock = clock
        self._condition = threading.Condition()
        self._wakeups = 0
        self._clock.add_listener(self._notify_from_clock)

    def __enter__(self) -> Self:
        self._condition.__enter__()
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        self._condition.__exit__(exc_type, exc, tb)

    def wait(self, timeout: float | None = None) -> bool:
        deadline = None if timeout is None else self._clock.monotonic() + timeout
        while True:
            if self._wakeups > 0:
                self._wakeups -= 1
                return True
            if deadline is not None and self._clock.monotonic() >= deadline:
                return True
            self._condition.wait(timeout=0.1)

    def notify_all(self) -> None:
        self._wakeups += 1
        self._condition.notify_all()

    def _notify_from_clock(self) -> None:
        with self._condition:
            self._condition.notify_all()


def _load_startup_manifest(
    backend: KubernetesWorkerBackend,
    *,
    worker_key: str,
) -> dict[str, object]:
    manifest_path = sandbox_startup_manifest_path(
        backend.storage_root / f"workers/{worker_dir_name(worker_key)}",
    )
    return json.loads(manifest_path.read_text(encoding="utf-8"))


class _FakeApiError(Exception):
    def __init__(self, status: int) -> None:
        super().__init__(status)
        self.status = status


_MAPPING_KEYS = {"annotations", "labels", "matchLabels", "selector", "stringData", "data"}


def _encoded_secret_data(values: dict[str, str]) -> dict[str, str]:
    return {key: base64.b64encode(value.encode("utf-8")).decode("ascii") for key, value in values.items()}


def _to_namespace(value: object, *, key: str | None = None) -> object:
    if isinstance(value, dict):
        if key in _MAPPING_KEYS:
            return deepcopy(value)
        return SimpleNamespace(**{item_key: _to_namespace(item, key=item_key) for item_key, item in value.items()})
    if isinstance(value, list):
        return [_to_namespace(item) for item in value]
    return value


class _FakeAppsApi:
    def __init__(self) -> None:
        self.deployments: dict[str, object] = {}
        self.read_names: list[str] = []
        self.created_bodies: list[dict[str, object]] = []
        self.patched_bodies: list[tuple[str, dict[str, object]]] = []
        self.deleted_names: list[str] = []
        self.list_label_selectors: list[str] = []
        self.raw_list_count = 0
        self.delete_read_lag_by_name: dict[str, int] = {}
        self._active_delete_read_lag_by_name: dict[str, int] = {}

    def read_namespaced_deployment(self, name: str, namespace: str) -> object:
        _ = namespace
        self.read_names.append(name)
        deployment = self.deployments.get(name)
        if deployment is None:
            raise _FakeApiError(404)
        remaining_delete_reads = self._active_delete_read_lag_by_name.get(name)
        if remaining_delete_reads is not None:
            if remaining_delete_reads <= 0:
                self._active_delete_read_lag_by_name.pop(name, None)
                self.deployments.pop(name, None)
                raise _FakeApiError(404)
            self._active_delete_read_lag_by_name[name] = remaining_delete_reads - 1
        return deployment

    def create_namespaced_deployment(self, namespace: str, body: dict[str, object]) -> object:
        _ = namespace
        self.created_bodies.append(body)
        deployment = _to_namespace(body)
        deployment.metadata.generation = 1
        deployment.metadata.uid = f"{deployment.metadata.name}-uid"
        deployment.status = SimpleNamespace(ready_replicas=body["spec"]["replicas"], observed_generation=1)
        self._active_delete_read_lag_by_name.pop(deployment.metadata.name, None)
        self.deployments[deployment.metadata.name] = deployment
        return deployment

    def patch_namespaced_deployment(self, name: str, namespace: str, body: dict[str, object]) -> object:
        _ = namespace
        self.patched_bodies.append((name, body))
        deployment = self.deployments.get(name)
        if deployment is None:
            raise _FakeApiError(404)
        metadata = body.get("metadata")
        if isinstance(metadata, dict):
            annotations = metadata.get("annotations")
            if isinstance(annotations, dict):
                deployment.metadata.annotations = {
                    **dict(deployment.metadata.annotations or {}),
                    **annotations,
                }
        spec = body.get("spec")
        if isinstance(spec, dict) and "replicas" in spec:
            deployment.spec.replicas = spec["replicas"]
            deployment.status.ready_replicas = spec["replicas"]
        deployment.metadata.generation += 1
        deployment.status.observed_generation = deployment.metadata.generation
        return deployment

    def delete_namespaced_deployment(self, name: str, namespace: str) -> None:
        _ = namespace
        self.deleted_names.append(name)
        if self.delete_read_lag_by_name.get(name, 0) > 0:
            self._active_delete_read_lag_by_name[name] = self.delete_read_lag_by_name[name]
            return
        self.deployments.pop(name, None)

    def list_namespaced_deployment(
        self,
        namespace: str,
        *,
        label_selector: str,
        _preload_content: bool = True,
    ) -> object:
        _ = namespace
        self.list_label_selectors.append(label_selector)
        selectors = {}
        for expression in filter(None, (part.strip() for part in label_selector.split(","))):
            key, sep, value = expression.partition("=")
            if not sep:
                continue
            selectors[key] = value

        def matches_selector(deployment: object) -> bool:
            labels = deployment.metadata.labels
            return all(labels.get(key) == value for key, value in selectors.items())

        deployments = [deployment for deployment in self.deployments.values() if matches_selector(deployment)]
        if _preload_content:
            return SimpleNamespace(items=deployments)
        self.raw_list_count += 1
        payload = {
            "items": [
                {
                    "metadata": {
                        "name": deployment.metadata.name,
                        "annotations": deployment.metadata.annotations,
                        "labels": deployment.metadata.labels,
                        "generation": deployment.metadata.generation,
                        "uid": deployment.metadata.uid,
                    },
                    "spec": {"replicas": deployment.spec.replicas},
                    "status": {
                        "readyReplicas": deployment.status.ready_replicas,
                        "observedGeneration": deployment.status.observed_generation,
                    },
                }
                for deployment in deployments
            ],
        }
        return _FakeRawResponse(json.dumps(payload).encode())


class _FakeRawResponse:
    def __init__(self, data: bytes) -> None:
        self.data = data
        self.released = False

    def release_conn(self) -> None:
        self.released = True


class _FakeCoreApi:
    def __init__(self) -> None:
        self.services: dict[str, object] = {}
        self.secrets: dict[str, object] = {}
        self.pods: dict[str, object] = {}
        self.read_service_names: list[str] = []
        self.read_secret_names: list[str] = []
        self.created_bodies: list[dict[str, object]] = []
        self.created_secret_bodies: list[dict[str, object]] = []
        self.patched_secret_bodies: list[tuple[str, object]] = []
        self.deleted_secret_names: list[str] = []
        self.api_client = _FakeApiClient(self)

    def read_namespaced_service(self, name: str, namespace: str) -> object:
        _ = namespace
        self.read_service_names.append(name)
        service = self.services.get(name)
        if service is None:
            raise _FakeApiError(404)
        return service

    def create_namespaced_service(self, namespace: str, body: dict[str, object]) -> object:
        _ = namespace
        self.created_bodies.append(body)
        service = _to_namespace(body)
        self.services[service.metadata.name] = service
        return service

    def patch_namespaced_service(self, name: str, namespace: str, body: dict[str, object]) -> object:
        _ = namespace
        service = _to_namespace(body)
        self.services[name] = service
        return service

    def delete_namespaced_service(self, name: str, namespace: str) -> None:
        _ = namespace
        self.services.pop(name, None)

    def read_namespaced_secret(self, name: str, namespace: str) -> object:
        _ = namespace
        self.read_secret_names.append(name)
        secret = self.secrets.get(name)
        if secret is None:
            raise _FakeApiError(404)
        return secret

    def create_namespaced_secret(self, namespace: str, body: dict[str, object]) -> object:
        _ = namespace
        self.created_secret_bodies.append(body)
        string_data = body.get("stringData")
        data = body.get("data")
        secret = _to_namespace(body)
        secret.data = _encoded_secret_data(string_data) if isinstance(string_data, dict) else {}
        if isinstance(data, dict):
            secret.data.update(data)
        secret.stringData = {}
        self.secrets[secret.metadata.name] = secret
        return secret

    def patch_namespaced_secret(self, name: str, namespace: str, body: dict[str, object], **kwargs: object) -> object:
        del namespace, kwargs
        self.patched_secret_bodies.append((name, body))
        secret = self.secrets.get(name)
        if secret is None:
            raise _FakeApiError(404)
        string_data = body.get("stringData")
        if isinstance(string_data, dict):
            current_data = dict(secret.data or {})
            current_data.update(_encoded_secret_data(string_data))
            secret.data = current_data
            secret.stringData = {}
        data = body.get("data")
        if isinstance(data, dict):
            current_data = dict(secret.data or {})
            for key, value in data.items():
                if value is None:
                    current_data.pop(key, None)
                else:
                    current_data[key] = value
            secret.data = current_data
        return secret

    def delete_namespaced_secret(self, name: str, namespace: str) -> None:
        _ = namespace
        self.deleted_secret_names.append(name)
        self.secrets.pop(name, None)

    def read_namespaced_pod(self, name: str, namespace: str) -> object:
        _ = namespace
        pod = self.pods.get(name)
        if pod is None:
            raise _FakeApiError(404)
        return pod


class _FakeApiClient:
    def __init__(self, core_api: _FakeCoreApi) -> None:
        self._core_api = core_api

    def select_header_accept(self, _content_types: list[str]) -> str:
        return "application/json"

    def call_api(
        self,
        _resource_path: str,
        _method: str,
        path_params: dict[str, str],
        _query_params: list[object],
        header_params: dict[str, str],
        *,
        body: object,
        **_kwargs: object,
    ) -> object:
        assert header_params["Content-Type"] == "application/merge-patch+json"
        name = path_params["name"]
        namespace = path_params["namespace"]
        return self._core_api.patch_namespaced_secret(name, namespace, body)


def test_kubernetes_deployment_snapshot_tolerates_missing_status() -> None:
    """A Deployment read before the controller writes status is unready, not fatal."""
    payload = {
        "metadata": {"name": "mindroom-worker", "labels": {}},
        "spec": {"replicas": 1},
    }

    snapshot = kubernetes_resources_module._deployment_snapshot(payload)

    assert snapshot.status.ready_replicas is None
    assert snapshot.status.observed_generation is None


def test_kubernetes_deployment_snapshot_tolerates_missing_labels() -> None:
    """Labels are carried but never read; their absence must not fail the pass."""
    payload = {
        "metadata": {"name": "mindroom-worker"},
        "spec": {"replicas": 1},
        "status": {},
    }

    assert kubernetes_resources_module._deployment_snapshot(payload).metadata.labels == {}


def test_kubernetes_deployment_snapshot_rejects_non_object_status() -> None:
    """A structurally wrong status is still a malformed payload."""
    payload = {
        "metadata": {"name": "mindroom-worker", "labels": {}},
        "spec": {"replicas": 1},
        "status": "ready",
    }

    with pytest.raises(WorkerBackendError, match="non-object status"):
        kubernetes_resources_module._deployment_snapshot(payload)


def _backend(
    *,
    idle_timeout_seconds: float = 60.0,
    worker_port: int = 8766,
    storage_subpath_prefix: str = "workers",
    storage_mount_path: str = "/app/worker",
    config_map_name: str | None = "mindroom-config",
    worker_config_path: str = "/app/config.yaml",
    node_name: str | None = None,
    colocate_with_control_plane_node: bool = False,
    name_prefix: str = "mindroom-worker",
    owner_deployment_name: str | None = None,
    runtime_paths: RuntimePaths | None = None,
    tool_validation_snapshot: dict[str, dict[str, object]] | None = None,
    worker_grantable_credentials: frozenset[str] = DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
    resource_requests: dict[str, str] | None = None,
    resource_limits: dict[str, str] | None = None,
    extra_env: dict[str, str] | None = None,
    extra_annotations: dict[str, str] | None = None,
    enable_service_links: bool = False,
    auth_secret_name: str | None = None,
    reconcile_pod_templates: bool = True,
    agent_vault: KubernetesAgentVaultConfig | None = None,
    config_snapshot: dict[str, object] | None = None,
) -> tuple[KubernetesWorkerBackend, _FakeAppsApi, _FakeCoreApi]:
    config = KubernetesWorkerBackendConfig(
        namespace="chat",
        image="ghcr.io/mindroom-ai/mindroom:latest",
        image_pull_policy="IfNotPresent",
        worker_port=worker_port,
        service_account_name="mindroom-worker",
        storage_pvc_name="mindroom-storage",
        storage_mount_path=storage_mount_path,
        storage_subpath_prefix=storage_subpath_prefix,
        config_map_name=config_map_name,
        config_key="config.yaml",
        config_path=worker_config_path,
        idle_timeout_seconds=idle_timeout_seconds,
        ready_timeout_seconds=5.0,
        name_prefix=name_prefix,
        node_name=node_name,
        colocate_with_control_plane_node=colocate_with_control_plane_node,
        extra_env=extra_env or {},
        extra_labels={"mindroom.ai/tenant": "test"},
        extra_annotations=extra_annotations or {},
        owner_deployment_name=owner_deployment_name,
        resource_requests=resource_requests if resource_requests is not None else {"memory": "256Mi", "cpu": "100m"},
        resource_limits=resource_limits if resource_limits is not None else {"memory": "1Gi", "cpu": "500m"},
        enable_service_links=enable_service_links,
        auth_secret_name=auth_secret_name,
        reconcile_pod_templates=reconcile_pod_templates,
        agent_vault=agent_vault,
    )
    resolved_runtime_paths = runtime_paths or resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=Path("mindroom-test-storage").resolve(),
    )
    if config_snapshot is None:
        try:
            loaded_config_snapshot, _source_files = load_yaml_config_source(resolved_runtime_paths.config_path)
        except OSError:
            loaded_config_snapshot = {}
        config_snapshot = loaded_config_snapshot if isinstance(loaded_config_snapshot, dict) else {}
    backend = KubernetesWorkerBackend(
        runtime_paths=resolved_runtime_paths,
        config=config,
        auth_token=_TEST_AUTH_TOKEN,
        storage_root=resolved_runtime_paths.storage_root,
        tool_validation_snapshot=tool_validation_snapshot or deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT),
        config_snapshot=config_snapshot,
        worker_grantable_credentials=worker_grantable_credentials,
    )
    apps_api = _FakeAppsApi()
    core_api = _FakeCoreApi()
    if auth_secret_name is not None:
        core_api.secrets[auth_secret_name] = SimpleNamespace(
            metadata=SimpleNamespace(
                name=auth_secret_name,
                annotations={},
                labels={},
                generation=1,
                uid=f"{auth_secret_name}-uid",
            ),
            stringData={},
            data={},
        )
    backend._resources.apps_api = apps_api
    backend._resources.core_api = core_api
    backend._resources.api_exception_cls = _FakeApiError
    if owner_deployment_name is not None:
        apps_api.deployments[owner_deployment_name] = SimpleNamespace(
            metadata=SimpleNamespace(
                name=owner_deployment_name,
                annotations={},
                labels={},
                generation=1,
                uid=f"{owner_deployment_name}-uid",
            ),
            spec=SimpleNamespace(replicas=1),
            status=SimpleNamespace(ready_replicas=1, observed_generation=1),
        )
    return backend, apps_api, core_api


def _install_real_elapsed_wait_for_ready(
    backend: KubernetesWorkerBackend,
    *,
    ready_after_seconds: float,
    ready_gate: threading.Event | None = None,
    poll_interval_seconds: float = 0.01,
    monotonic: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
    on_iteration: Callable[[float], None] | None = None,
) -> None:
    object.__setattr__(
        backend.config,
        "ready_timeout_seconds",
        max(backend.config.ready_timeout_seconds, ready_after_seconds + 1.0),
    )

    def _ready(
        self: object,
        deployment_name: str,
        *,
        timeout_seconds: float,
        deployment_ready_fn: object,
        on_poll_tick: Callable[[float], None] | None = None,
    ) -> object:
        del deployment_ready_fn
        started_at = monotonic()
        deadline = started_at + timeout_seconds
        while True:
            elapsed_seconds = monotonic() - started_at
            if on_iteration is not None:
                on_iteration(elapsed_seconds)
            if elapsed_seconds >= ready_after_seconds and (ready_gate is None or ready_gate.is_set()):
                deployment = self.read_deployment(deployment_name)
                assert deployment is not None
                return deployment
            assert monotonic() < deadline
            if on_poll_tick is not None:
                on_poll_tick(elapsed_seconds)
            sleep(poll_interval_seconds)

    backend._resources.wait_for_ready = MethodType(_ready, backend._resources)


def test_kubernetes_backend_ensures_worker_service_deployment_and_auth_secret(tmp_path: Path) -> None:  # noqa: PLR0915
    """Ensuring one worker should create runtime resources on shared storage."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        owner_deployment_name="mindroom-demo",
    )
    worker_key = _TEST_SCOPED_WORKER_KEY_A

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

    assert handle.worker_key == worker_key
    assert handle.auth_token == worker_auth_token(_TEST_AUTH_TOKEN, worker_key)
    assert handle.auth_token != _TEST_AUTH_TOKEN
    assert handle.backend_name == "kubernetes"
    assert handle.endpoint.endswith("/api/sandbox-runner/execute")
    assert handle.debug_metadata["namespace"] == "chat"
    assert handle.debug_metadata["state_subpath"] == f"workers/{worker_dir_name(worker_key)}"
    assert handle.debug_metadata["service_name"] == handle.worker_id
    assert handle.status == "ready"

    assert len(core_api.created_bodies) == 1
    assert len(core_api.created_secret_bodies) == 1
    assert len(apps_api.created_bodies) == 1
    auth_secret = core_api.created_secret_bodies[0]
    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_by_name = {env["name"]: env for env in container["env"]}
    env_values = {env["name"]: env.get("value") for env in container["env"]}
    env_names = set(env_values)
    token_env = env_by_name["MINDROOM_SANDBOX_PROXY_TOKEN"]
    assert "MINDROOM_SANDBOX_DEDICATED_WORKER_KEY" in env_names
    assert "MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT" in env_names
    assert "MINDROOM_STORAGE_PATH" in env_names
    assert "MINDROOM_SANDBOX_STARTUP_MANIFEST_PATH" in env_names
    assert "MINDROOM_SANDBOX_SHARED_STORAGE_ROOT" not in env_names
    assert "VIRTUAL_ENV" in env_names
    assert "PATH" in env_names
    assert "MINDROOM_SHARED_CREDENTIALS_PATH" in env_names
    assert token_env == {
        "name": "MINDROOM_SANDBOX_PROXY_TOKEN",
        "valueFrom": {
            "secretKeyRef": {
                "name": handle.worker_id,
                "key": "MINDROOM_SANDBOX_PROXY_TOKEN",
            },
        },
    }
    assert auth_secret["metadata"]["name"] == handle.worker_id
    assert auth_secret["stringData"] == {"MINDROOM_SANDBOX_PROXY_TOKEN": handle.auth_token}
    assert handle.auth_token not in json.dumps(deployment)
    assert deployment["spec"]["template"]["metadata"]["annotations"][_ANNOTATION_RUNNER_TOKEN_HASH]
    assert deployment["spec"]["template"]["metadata"]["annotations"][_ANNOTATION_RUNNER_TOKEN_HASH] != handle.auth_token
    assert env_values["MINDROOM_SANDBOX_RUNNER_EXECUTION_MODE"] == "forkserver"
    assert env_values["MINDROOM_SANDBOX_RUNNER_PORT"] == "8766"
    manifest_path = env_values["MINDROOM_SANDBOX_STARTUP_MANIFEST_PATH"]
    assert manifest_path is not None
    startup_manifest = _load_startup_manifest(backend, worker_key=worker_key)
    validation_snapshot = startup_manifest["tool_validation_snapshot"]
    assert validation_snapshot == _TEST_TOOL_VALIDATION_SNAPSHOT
    expected_dedicated_root = f"/app/worker/workers/{worker_dir_name(worker_key)}"
    assert manifest_path == f"{expected_dedicated_root}/.runtime/startup_manifest.json"
    committed_runtime = deserialize_runtime_paths(startup_manifest["runtime_paths"])
    assert env_values["MINDROOM_STORAGE_PATH"] == expected_dedicated_root
    assert env_values["MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT"] == expected_dedicated_root
    assert env_values["HOME"] == expected_dedicated_root
    assert env_values["VIRTUAL_ENV"] == f"{expected_dedicated_root}/venv"
    assert env_values["PATH"].startswith(f"{expected_dedicated_root}/venv/bin:")
    assert env_values["MINDROOM_SHARED_CREDENTIALS_PATH"] == f"{expected_dedicated_root}/.shared_credentials"
    assert committed_runtime.storage_root == Path(expected_dedicated_root)
    assert committed_runtime.env_value("MINDROOM_SANDBOX_DEDICATED_WORKER_KEY") == worker_key
    assert committed_runtime.env_value("MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT") == expected_dedicated_root
    assert (
        deployment["spec"]["template"]["spec"]["volumes"][0]["persistentVolumeClaim"]["claimName"] == "mindroom-storage"
    )
    assert deployment["metadata"]["labels"]["mindroom.ai/tenant"] == "test"
    assert deployment["metadata"]["ownerReferences"] == [
        {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "name": "mindroom-demo",
            "uid": "mindroom-demo-uid",
            "controller": False,
            "blockOwnerDeletion": False,
        },
    ]
    assert core_api.created_bodies[0]["metadata"]["ownerReferences"] == [
        {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "name": "mindroom-demo",
            "uid": "mindroom-demo-uid",
            "controller": False,
            "blockOwnerDeletion": False,
        },
    ]
    assert auth_secret["metadata"]["ownerReferences"] == core_api.created_bodies[0]["metadata"]["ownerReferences"]
    template_annotations = deployment["spec"]["template"]["metadata"]["annotations"]
    assert set(template_annotations) == {
        _ANNOTATION_RUNNER_TOKEN_HASH,
        _ANNOTATION_STARTUP_MANIFEST_HASH,
        ANNOTATION_WORKER_KEY,
    }
    assert template_annotations[ANNOTATION_WORKER_KEY] == worker_key
    assert template_annotations[_ANNOTATION_RUNNER_TOKEN_HASH] == kubernetes_resources_module._worker_auth_token_hash(
        _TEST_AUTH_TOKEN,
        worker_key,
    )
    assert template_annotations[_ANNOTATION_STARTUP_MANIFEST_HASH] == startup_manifest_sha256(
        committed_runtime,
        tool_validation_snapshot=_TEST_TOOL_VALIDATION_SNAPSHOT,
        public_runtime=True,
    )
    assert deployment["spec"]["template"]["spec"]["securityContext"] == {
        "runAsUser": 1000,
        "runAsGroup": 1000,
        "fsGroup": 1000,
        "runAsNonRoot": True,
        "fsGroupChangePolicy": "OnRootMismatch",
        "seccompProfile": {"type": "RuntimeDefault"},
    }
    assert container["securityContext"] == {
        "allowPrivilegeEscalation": False,
        "capabilities": {"drop": ["ALL"]},
    }
    assert container["resources"]["requests"] == {"memory": "256Mi", "cpu": "100m"}
    assert container["resources"]["limits"] == {"memory": "1Gi", "cpu": "500m"}
    assert deployment["spec"]["template"]["spec"]["automountServiceAccountToken"] is False
    assert deployment["spec"]["template"]["spec"]["enableServiceLinks"] is False
    assert container["startupProbe"] == {
        "httpGet": {"path": "/healthz", "port": "api"},
        "periodSeconds": 5,
        "failureThreshold": 60,
    }


def test_kubernetes_worker_startup_manifest_omits_credentials_encryption_key(tmp_path: Path) -> None:
    """Worker manifests should not persist credential encryption key material beside worker state."""
    encryption_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    worker_key = _TEST_SCOPED_WORKER_KEY_A

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

    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_by_name = {env["name"]: env for env in container["env"]}
    startup_manifest = _load_startup_manifest(backend, worker_key=worker_key)
    committed_runtime = deserialize_runtime_paths(startup_manifest["runtime_paths"])

    assert env_by_name[CREDENTIALS_ENCRYPTION_KEY_ENV] == {
        "name": CREDENTIALS_ENCRYPTION_KEY_ENV,
        "valueFrom": {
            "secretKeyRef": {
                "name": handle.worker_id,
                "key": CREDENTIALS_ENCRYPTION_KEY_ENV,
            },
        },
    }
    assert committed_runtime.env_value(CREDENTIALS_ENCRYPTION_KEY_ENV) is None
    assert core_api.created_secret_bodies[0]["stringData"][CREDENTIALS_ENCRYPTION_KEY_ENV] == encryption_key
    assert encryption_key not in json.dumps(deployment)
    assert encryption_key not in json.dumps(startup_manifest)


def test_kubernetes_worker_credentials_encryption_key_uses_runtime_source_not_extra_env(tmp_path: Path) -> None:
    """Worker credential encryption should use the same runtime key source as CredentialsManager."""
    runtime_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    extra_env_key = base64.urlsafe_b64encode(b"1" * 32).decode("ascii")
    carrier_env = json.dumps({CREDENTIALS_ENCRYPTION_KEY_ENV: extra_env_key})
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={
            CREDENTIALS_ENCRYPTION_KEY_ENV: runtime_key,
            "MINDROOM_KUBERNETES_WORKER_ENV_JSON": carrier_env,
        },
    )
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        extra_env={CREDENTIALS_ENCRYPTION_KEY_ENV: extra_env_key},
    )

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

    deployment = apps_api.created_bodies[0]
    startup_manifest = _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)
    assert core_api.created_secret_bodies[0]["stringData"][CREDENTIALS_ENCRYPTION_KEY_ENV] == runtime_key
    assert extra_env_key not in json.dumps(deployment)
    assert extra_env_key not in json.dumps(core_api.created_secret_bodies)
    assert extra_env_key not in json.dumps(startup_manifest)
    assert "MINDROOM_KUBERNETES_WORKER_ENV_JSON" not in json.dumps(startup_manifest)


def test_kubernetes_worker_credentials_encryption_key_rotation_changes_template_hash(tmp_path: Path) -> None:
    """Rotating the credential encryption key should restart workers without exposing the key."""
    first_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    second_key = base64.urlsafe_b64encode(b"1" * 32).decode("ascii")

    def deployment_for_key(encryption_key: str) -> dict[str, object]:
        runtime_paths = resolve_primary_runtime_paths(
            config_path=Path("config.yaml"),
            storage_path=tmp_path / f"mindroom-test-storage-{encryption_key[:4]}",
            process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
        )
        backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
        return apps_api.created_bodies[0]

    first_deployment = deployment_for_key(first_key)
    second_deployment = deployment_for_key(second_key)
    first_template_annotations = first_deployment["spec"]["template"]["metadata"]["annotations"]
    second_template_annotations = second_deployment["spec"]["template"]["metadata"]["annotations"]

    assert (
        first_template_annotations[_ANNOTATION_CREDENTIALS_ENCRYPTION_KEY_HASH]
        == hashlib.sha256(
            first_key.encode("utf-8"),
        ).hexdigest()
    )
    assert (
        second_template_annotations[_ANNOTATION_CREDENTIALS_ENCRYPTION_KEY_HASH]
        == hashlib.sha256(
            second_key.encode("utf-8"),
        ).hexdigest()
    )
    assert (
        first_template_annotations[_ANNOTATION_CREDENTIALS_ENCRYPTION_KEY_HASH]
        != (second_template_annotations[_ANNOTATION_CREDENTIALS_ENCRYPTION_KEY_HASH])
    )
    assert (
        first_deployment["metadata"]["annotations"][_ANNOTATION_TEMPLATE_HASH]
        != (second_deployment["metadata"]["annotations"][_ANNOTATION_TEMPLATE_HASH])
    )
    assert first_key not in json.dumps(first_deployment)
    assert second_key not in json.dumps(second_deployment)


def test_kubernetes_backend_config_signature_changes_with_credentials_encryption_key(tmp_path: Path) -> None:
    """Credential encryption key changes should invalidate cached Kubernetes managers without storing the key."""
    base_env = {
        "MINDROOM_KUBERNETES_WORKER_IMAGE": "test-image",
        "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME": "test-pvc",
    }
    first_key = base64.urlsafe_b64encode(b"1" * 32).decode("ascii")
    second_key = base64.urlsafe_b64encode(b"2" * 32).decode("ascii")
    first_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={**base_env, CREDENTIALS_ENCRYPTION_KEY_ENV: first_key},
    )
    second_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={**base_env, CREDENTIALS_ENCRYPTION_KEY_ENV: second_key},
    )
    disabled_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env=base_env,
    )

    first_signature = kubernetes_backend_config_signature(
        first_runtime_paths,
        auth_token=_TEST_AUTH_TOKEN,
        storage_root=first_runtime_paths.storage_root,
    )
    second_signature = kubernetes_backend_config_signature(
        second_runtime_paths,
        auth_token=_TEST_AUTH_TOKEN,
        storage_root=second_runtime_paths.storage_root,
    )
    disabled_signature = kubernetes_backend_config_signature(
        disabled_runtime_paths,
        auth_token=_TEST_AUTH_TOKEN,
        storage_root=disabled_runtime_paths.storage_root,
    )

    first_key_hash = hashlib.sha256(first_key.encode("utf-8")).hexdigest()
    assert first_signature != second_signature
    assert first_signature != disabled_signature
    assert first_key_hash in first_signature
    assert first_key not in "\n".join(first_signature)
    assert second_key not in "\n".join(second_signature)


def test_kubernetes_backend_can_use_one_precreated_auth_secret(tmp_path: Path) -> None:
    """Shared-namespace charts should need RBAC only for one tenant-owned Secret."""
    encryption_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths, auth_secret_name=auth_secret_name)
    worker_key = _TEST_SCOPED_WORKER_KEY_A

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

    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_by_name = {env["name"]: env for env in container["env"]}
    assert env_by_name["MINDROOM_SANDBOX_PROXY_TOKEN"] == {
        "name": "MINDROOM_SANDBOX_PROXY_TOKEN",
        "valueFrom": {
            "secretKeyRef": {
                "name": auth_secret_name,
                "key": handle.worker_id,
            },
        },
    }
    assert env_by_name[CREDENTIALS_ENCRYPTION_KEY_ENV] == {
        "name": CREDENTIALS_ENCRYPTION_KEY_ENV,
        "valueFrom": {
            "secretKeyRef": {
                "name": auth_secret_name,
                "key": f"{handle.worker_id}.credentials-encryption-key",
            },
        },
    }
    assert encryption_key not in json.dumps(deployment)
    assert core_api.created_secret_bodies == []
    expected_secret_data = _encoded_secret_data(
        {
            handle.worker_id: handle.auth_token,
            f"{handle.worker_id}.credentials-encryption-key": encryption_key,
        },
    )
    assert core_api.patched_secret_bodies[0] == (
        auth_secret_name,
        {"data": expected_secret_data},
    )
    assert core_api.secrets[auth_secret_name].data == expected_secret_data


def test_kubernetes_backend_cleanup_removes_only_own_key_from_tenant_auth_secret(tmp_path: Path) -> None:
    """Cleaning up one idle worker should null out only its key in the shared tenant Secret."""
    encryption_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, _apps_api, core_api = _backend(runtime_paths=runtime_paths, auth_secret_name=auth_secret_name)
    other_worker_id = "mindroom-worker-other"
    core_api.secrets[auth_secret_name].data = {other_worker_id: "preexisting"}
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    backend.cleanup_idle_workers(now=80.0)

    assert core_api.deleted_secret_names == []
    delete_patch = (
        auth_secret_name,
        {"data": {handle.worker_id: None, f"{handle.worker_id}.credentials-encryption-key": None}},
    )
    assert delete_patch in core_api.patched_secret_bodies
    tenant_secret = core_api.secrets[auth_secret_name]
    assert handle.worker_id not in tenant_secret.data
    assert f"{handle.worker_id}.credentials-encryption-key" not in tenant_secret.data
    assert tenant_secret.data[other_worker_id] == "preexisting"


def test_kubernetes_backend_reapply_without_encryption_removes_worker_secret_key(tmp_path: Path) -> None:
    """Reapplying a worker Secret after disabling encryption should remove stale key data."""
    encryption_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    encrypted_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
    )
    backend, _apps_api, core_api = _backend(runtime_paths=encrypted_runtime_paths)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    assert CREDENTIALS_ENCRYPTION_KEY_ENV in core_api.secrets[handle.worker_id].data

    unencrypted_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend.runtime_paths = unencrypted_runtime_paths
    backend._resources.runtime_paths = unencrypted_runtime_paths

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

    assert CREDENTIALS_ENCRYPTION_KEY_ENV not in core_api.secrets[handle.worker_id].data
    assert any(
        name == handle.worker_id and body.get("data", {}).get(CREDENTIALS_ENCRYPTION_KEY_ENV) is None
        for name, body in core_api.patched_secret_bodies
    )


def test_kubernetes_backend_reapply_without_encryption_removes_shared_secret_key(tmp_path: Path) -> None:
    """Reapplying a shared Secret entry after disabling encryption should remove stale worker key data."""
    encryption_key = base64.urlsafe_b64encode(b"0" * 32).decode("ascii")
    encrypted_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, _apps_api, core_api = _backend(runtime_paths=encrypted_runtime_paths, auth_secret_name=auth_secret_name)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    encryption_secret_key = f"{handle.worker_id}.credentials-encryption-key"
    assert encryption_secret_key in core_api.secrets[auth_secret_name].data

    unencrypted_runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend.runtime_paths = unencrypted_runtime_paths
    backend._resources.runtime_paths = unencrypted_runtime_paths

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

    assert encryption_secret_key not in core_api.secrets[auth_secret_name].data
    assert any(
        name == auth_secret_name and body.get("data", {}).get(encryption_secret_key) is None
        for name, body in core_api.patched_secret_bodies
    )


def test_kubernetes_backend_startup_failure_removes_key_from_tenant_auth_secret(tmp_path: Path) -> None:
    """Failing to apply the worker Deployment should release the tenant-Secret key."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, _apps_api, core_api = _backend(runtime_paths=runtime_paths, auth_secret_name=auth_secret_name)

    def apply_deployment_with_failure(**_kwargs: object) -> object:
        msg = "deployment apply failed"
        raise WorkerBackendError(msg)

    backend._resources.apply_deployment = apply_deployment_with_failure  # type: ignore[method-assign]
    worker_id = backend._worker_id(_TEST_SCOPED_WORKER_KEY_A)

    with pytest.raises(WorkerBackendError):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    delete_patch = (
        auth_secret_name,
        {"data": {worker_id: None, f"{worker_id}.credentials-encryption-key": None}},
    )
    assert delete_patch in core_api.patched_secret_bodies
    assert worker_id not in core_api.secrets[auth_secret_name].data


def test_kubernetes_backend_recreate_failure_removes_orphaned_tenant_auth_secret_key(tmp_path: Path) -> None:
    """If recreate deletes the worker Deployment before failing, the shared Secret key must be released."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        auth_secret_name=auth_secret_name,
        idle_timeout_seconds=600.0,
    )
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    deployment = apps_api.deployments[handle.worker_id]
    deployment.metadata.annotations[_ANNOTATION_TEMPLATE_HASH] = "stale"

    def recreate_with_failure(deployment_name: str, _manifest: dict[str, object], *, timeout_seconds: float) -> None:
        _ = timeout_seconds
        apps_api.delete_namespaced_deployment(deployment_name, backend.config.namespace)
        msg = "deployment recreate failed"
        raise WorkerBackendError(msg)

    backend._resources._recreate_deployment = recreate_with_failure  # type: ignore[method-assign]

    with pytest.raises(WorkerBackendError, match="deployment recreate failed"):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=310.0)

    assert handle.worker_id not in apps_api.deployments
    assert handle.worker_id not in core_api.secrets[auth_secret_name].stringData
    assert handle.worker_id not in core_api.secrets[auth_secret_name].data


def test_kubernetes_backend_shared_auth_secret_cleanup_ignores_missing_secret(tmp_path: Path) -> None:
    """Cleanup should stay idempotent if the shared tenant auth Secret was already removed."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    auth_secret_name = "mindroom-worker-auth-demo"  # noqa: S105
    backend, _apps_api, core_api = _backend(runtime_paths=runtime_paths, auth_secret_name=auth_secret_name)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    core_api.secrets.pop(auth_secret_name)

    cleaned = backend.cleanup_idle_workers(now=80.0)

    assert [worker.worker_key for worker in cleaned] == [_TEST_SCOPED_WORKER_KEY_A]
    assert (
        auth_secret_name,
        {"data": {handle.worker_id: None, f"{handle.worker_id}.credentials-encryption-key": None}},
    ) in core_api.patched_secret_bodies


def test_kubernetes_backend_recreates_worker_when_startup_manifest_changes(tmp_path: Path) -> None:
    """Changing startup state should force a worker Deployment recreate."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    initial_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot["search"] = {
        "config_fields": ["engine"],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    }
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        owner_deployment_name="mindroom-demo",
        tool_validation_snapshot=initial_snapshot,
    )
    worker_key = _TEST_SCOPED_WORKER_KEY_A

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

    initial_deployment = apps_api.created_bodies[0]
    initial_manifest_hash = initial_deployment["spec"]["template"]["metadata"]["annotations"][
        _ANNOTATION_STARTUP_MANIFEST_HASH
    ]

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        owner_deployment_name="mindroom-demo",
        tool_validation_snapshot=updated_snapshot,
    )
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError

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

    assert apps_api.deleted_names == [handle.worker_id]
    assert len(apps_api.created_bodies) == 2
    updated_deployment = apps_api.created_bodies[1]
    updated_manifest_hash = updated_deployment["spec"]["template"]["metadata"]["annotations"][
        _ANNOTATION_STARTUP_MANIFEST_HASH
    ]
    assert updated_manifest_hash != initial_manifest_hash


def test_kubernetes_backend_commits_parent_runtime_env_into_worker_payload(tmp_path: Path) -> None:
    """Dedicated worker startup payloads should keep worker control env while denying ambient provider and arbitrary runtime env."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    credentials_path = tmp_path / "google-credentials.json"
    credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    storage_mount_path = tmp_path / "worker-storage"
    storage_mount_path.mkdir()
    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_NAMESPACE=alpha1234\n"
            "MATRIX_HOMESERVER=http://dotenv-hs\n"
            "MATRIX_SERVER_NAME=alpha.example\n"
            "MATRIX_ACCESS_TOKEN=matrix-access-secret\n"
            "MATRIX_REGISTRATION_TOKEN=matrix-registration-secret\n"
            "CUSTOMER_ID=tenant-123\n"
            "ACCOUNT_ID=account-456\n"
            f"GOOGLE_APPLICATION_CREDENTIALS={credentials_path}\n"
            "GOOGLE_CLOUD_PROJECT=demo-project\n"
            "GOOGLE_CLOUD_LOCATION=us-central1\n"
            "OPENAI_BASE_URL=http://example.invalid/v1\n"
            "CUSTOM_API_TOKEN=custom-secret\n"
            "ANTHROPIC_API_KEY=sk-secret\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        process_env={
            "MINDROOM_SANDBOX_CREDENTIAL_POLICY_JSON": '{"shell":["github"]}',
            "MINDROOM_SANDBOX_PROXY_TOKEN": "test-token",
            "MINDROOM_SANDBOX_PROXY_TOOLS": "*",
            "MINDROOM_SANDBOX_PROXY_URL": "http://runner.example.invalid",
            "MINDROOM_SANDBOX_SHARED_STORAGE_ROOT": str(tmp_path / "primary-shared-root"),
            "MINDROOM_LOCAL_CLIENT_SECRET": "client-secret",
        },
    )
    backend, _apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        storage_mount_path=str(storage_mount_path),
    )

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

    committed_runtime = deserialize_runtime_paths(
        _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)["runtime_paths"],
    )
    state_subpath = Path("workers") / worker_dir_name(_TEST_SCOPED_WORKER_KEY_A)
    local_credentials_path = runtime_paths.storage_root / state_subpath / ".runtime" / credentials_path.name

    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("MATRIX_ACCESS_TOKEN") is None
    assert committed_runtime.env_value("MATRIX_REGISTRATION_TOKEN") is None
    assert committed_runtime.env_value("CUSTOMER_ID") == "tenant-123"
    assert committed_runtime.env_value("ACCOUNT_ID") == "account-456"
    assert committed_runtime.env_value("GOOGLE_APPLICATION_CREDENTIALS") is None
    assert committed_runtime.env_value("GOOGLE_CLOUD_PROJECT") is None
    assert committed_runtime.env_value("GOOGLE_CLOUD_LOCATION") is None
    assert committed_runtime.env_value("OPENAI_BASE_URL") is None
    assert committed_runtime.env_value("CUSTOM_API_TOKEN") is None
    assert committed_runtime.env_value("ANTHROPIC_API_KEY") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_CREDENTIAL_POLICY_JSON") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_PROXY_TOKEN") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_PROXY_TOOLS") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_PROXY_URL") is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_SHARED_STORAGE_ROOT") is None
    assert committed_runtime.env_value("MINDROOM_LOCAL_CLIENT_SECRET") is None
    assert not local_credentials_path.exists()


def test_kubernetes_backend_uses_provided_validation_snapshot(tmp_path: Path) -> None:
    """Worker env should reflect the authoritative snapshot passed into the backend."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    custom_snapshot = {
        "worker_only": {
            "config_fields": [],
            "agent_override_fields": [],
            "authored_override_validator": "default",
            "runtime_loadable": False,
        },
    }
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        owner_deployment_name="mindroom-demo",
        tool_validation_snapshot=custom_snapshot,
    )

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

    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_values = {env["name"]: env.get("value") for env in container["env"]}

    assert env_values["MINDROOM_SANDBOX_STARTUP_MANIFEST_PATH"] is not None
    assert (
        _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)["tool_validation_snapshot"]
        == custom_snapshot
    )


def test_kubernetes_backend_drops_host_local_adc_path_when_not_mounted(tmp_path: Path) -> None:
    """Dedicated worker payloads must not serialize unusable host-local ADC paths."""
    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",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        process_env={"GOOGLE_APPLICATION_CREDENTIALS": "/host/path/adc.json"},
    )
    backend, _apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        storage_mount_path=str(tmp_path / "not-mounted-storage"),
    )

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

    committed_runtime = deserialize_runtime_paths(
        _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)["runtime_paths"],
    )

    assert committed_runtime.env_value("GOOGLE_APPLICATION_CREDENTIALS") is None


def test_kubernetes_backend_rejects_google_vertex_adc_worker_grant(tmp_path: Path) -> None:
    """Dedicated workers should reject google_vertex_adc instead of accepting a non-working grant."""
    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",
    )
    credentials_path = tmp_path / "adc.json"
    credentials_path.write_text('{"type":"service_account"}\n', encoding="utf-8")
    local_storage_root = tmp_path / "local-shared-storage"
    local_storage_root.mkdir()
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=local_storage_root,
        process_env={"GOOGLE_APPLICATION_CREDENTIALS": str(credentials_path)},
    )
    with pytest.raises(WorkerBackendError, match="google_vertex_adc"):
        _backend(
            runtime_paths=runtime_paths,
            storage_mount_path="/app/worker",
            worker_grantable_credentials=frozenset({"google_vertex_adc"}),
        )


def test_kubernetes_backend_preserves_primary_config_path_without_configmap(tmp_path: Path) -> None:
    """Dedicated worker payloads should keep the primary runtime config path when no ConfigMap is mounted."""
    config_path = tmp_path / "workspace-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")
    backend, _apps_api, _core_api = _backend(runtime_paths=runtime_paths, config_map_name=None)

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

    committed_runtime = deserialize_runtime_paths(
        _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)["runtime_paths"],
    )

    assert committed_runtime.config_path == config_path.resolve()


@pytest.mark.parametrize(
    ("config_relative_path", "worker_config_path", "expected_mount_path", "expected_subpath"),
    [
        (
            "content-bundles/team-config/agent-config.yaml",
            "/app/agent_data/content-bundles/team-config/agent-config.yaml",
            "/app/agent_data/content-bundles",
            "content-bundles",
        ),
        (
            "team-config/content/environments/prod/agent-config.yaml",
            "/app/agent_data/team-config/content/environments/prod/agent-config.yaml",
            "/app/agent_data/team-config",
            "team-config",
        ),
    ],
)
def test_kubernetes_backend_mounts_config_storage_subtree_without_configmap(
    tmp_path: Path,
    config_relative_path: str,
    worker_config_path: str,
    expected_mount_path: str,
    expected_subpath: str,
) -> None:
    """File-backed configs need bundle visibility without broadening worker state mounts."""
    config_path = tmp_path / "storage" / config_relative_path
    config_path.parent.mkdir(parents=True)
    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")
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        storage_mount_path="/app/agent_data",
        config_map_name=None,
        worker_config_path=worker_config_path,
    )

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

    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_by_name = {env["name"]: env for env in container["env"]}
    mount_paths = {mount["mountPath"]: mount for mount in container["volumeMounts"]}
    expected_worker_root = f"/app/agent_data/workers/{worker_dir_name(_TEST_SCOPED_WORKER_KEY_A)}"

    assert mount_paths[expected_mount_path] == {
        "name": "worker-storage",
        "mountPath": expected_mount_path,
        "subPath": expected_subpath,
        "readOnly": True,
    }
    assert "/app/agent_data" not in mount_paths
    assert mount_paths["/app/agent_data/agents/code"]["subPath"] == "agents/code"
    assert mount_paths[expected_worker_root]["subPath"] == f"workers/{worker_dir_name(_TEST_SCOPED_WORKER_KEY_A)}"
    assert not any(mount["name"] == "worker-config" for mount in container["volumeMounts"])
    assert deployment["spec"]["template"]["spec"]["volumes"] == [
        {"name": "worker-storage", "persistentVolumeClaim": {"claimName": "mindroom-storage"}},
    ]
    assert env_by_name["MINDROOM_CONFIG_PATH"]["value"] == worker_config_path


def test_primary_worker_backend_available_uses_runtime_env_values(tmp_path: Path) -> None:
    """Kubernetes 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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\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) == "kubernetes"
    assert runtime_paths.env_value("MINDROOM_KUBERNETES_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_kubernetes_backend_config_resource_envs_override_defaults(tmp_path: Path) -> None:
    """Resource request/limit env vars override the built-in defaults."""
    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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
            "MINDROOM_KUBERNETES_WORKER_MEMORY_REQUEST=2Gi\n"
            "MINDROOM_KUBERNETES_WORKER_MEMORY_LIMIT=8Gi\n"
            "MINDROOM_KUBERNETES_WORKER_CPU_REQUEST=500m\n"
            "MINDROOM_KUBERNETES_WORKER_CPU_LIMIT=2\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    config = KubernetesWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.resource_requests == {"memory": "2Gi", "cpu": "500m"}
    assert config.resource_limits == {"memory": "8Gi", "cpu": "2"}


def test_kubernetes_backend_config_resources_default_when_env_unset(tmp_path: Path) -> None:
    """Resource defaults kick in when no override env vars are present."""
    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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    config = KubernetesWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.resource_requests == {"memory": "256Mi", "cpu": "100m"}
    assert config.resource_limits == {"memory": "1Gi", "cpu": "500m"}
    assert config.enable_service_links is False


def test_kubernetes_backend_config_allows_service_links_override(tmp_path: Path) -> None:
    """Worker service-link env injection remains opt-in."""
    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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
            "MINDROOM_KUBERNETES_WORKER_ENABLE_SERVICE_LINKS=true\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    config = KubernetesWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.enable_service_links is True


def test_kubernetes_backend_renders_configured_resources_on_worker_container(tmp_path: Path) -> None:
    """Configured resource requests/limits land on the rendered worker container."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        resource_requests={"memory": "2Gi", "cpu": "500m"},
        resource_limits={"memory": "8Gi", "cpu": "2"},
    )

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

    container = apps_api.created_bodies[0]["spec"]["template"]["spec"]["containers"][0]
    assert container["resources"]["requests"] == {"memory": "2Gi", "cpu": "500m"}
    assert container["resources"]["limits"] == {"memory": "8Gi", "cpu": "2"}


def test_kubernetes_backend_renders_service_links_override_in_worker_template() -> None:
    """Generated worker pod templates should carry the configured service-link setting."""
    backend, apps_api, _core_api = _backend(enable_service_links=True)

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

    deployment = apps_api.created_bodies[0]
    assert deployment["spec"]["template"]["spec"]["enableServiceLinks"] is True


def test_kubernetes_backend_config_reads_worker_annotations_from_env(tmp_path: Path) -> None:
    """Worker pod annotations are configurable through runtime env."""
    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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
            'MINDROOM_KUBERNETES_WORKER_ANNOTATIONS_JSON={"cluster-autoscaler.kubernetes.io/safe-to-evict":"false"}\n'
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    config = KubernetesWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.extra_annotations == {
        "cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
    }


def test_kubernetes_backend_renders_configured_annotations_on_worker_pod_template(tmp_path: Path) -> None:
    """Configured annotations land on the rendered worker pod template."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        extra_annotations={
            "cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
            ANNOTATION_WORKER_KEY: "user-supplied-value",
            _ANNOTATION_STARTUP_MANIFEST_HASH: "user-supplied-value",
        },
    )

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

    annotations = apps_api.created_bodies[0]["spec"]["template"]["metadata"]["annotations"]
    startup_manifest = _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)
    committed_runtime = deserialize_runtime_paths(startup_manifest["runtime_paths"])
    assert annotations["cluster-autoscaler.kubernetes.io/safe-to-evict"] == "false"
    assert annotations[ANNOTATION_WORKER_KEY] == _TEST_SCOPED_WORKER_KEY_A
    assert annotations[_ANNOTATION_STARTUP_MANIFEST_HASH] == startup_manifest_sha256(
        committed_runtime,
        tool_validation_snapshot=_TEST_TOOL_VALIDATION_SNAPSHOT,
        public_runtime=True,
    )
    assert annotations[ANNOTATION_WORKER_KEY] != "user-supplied-value"
    assert annotations[_ANNOTATION_STARTUP_MANIFEST_HASH] != "user-supplied-value"


def test_kubernetes_backend_omits_backend_config_env_from_worker_env_and_manifest(tmp_path: Path) -> None:
    """Primary-side backend config carriers do not reach worker pods or startup manifests."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
        process_env={
            "MINDROOM_KUBERNETES_WORKER_ENV_JSON": json.dumps({"OPENAI_API_KEY": "nested-secret"}),
            "MINDROOM_KUBERNETES_WORKER_LABELS_JSON": "{}",
            "MINDROOM_KUBERNETES_WORKER_ANNOTATIONS_JSON": "{}",
            "MINDROOM_KUBERNETES_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME": "mindroom-storage",
            "MINDROOM_SHARED_CREDENTIALS_PATH": str(tmp_path / "primary-shared-credentials"),
            "MATRIX_HOMESERVER": "https://matrix.example.invalid",
        },
    )
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        extra_env={
            "HOME": "/unsafe/home",
            "MINDROOM_API_KEY": "runtime-api-key",
            "MINDROOM_CONFIG_PATH": "/unsafe/config.yaml",
            "MINDROOM_KUBERNETES_WORKER_ENV_JSON": json.dumps({"ANTHROPIC_API_KEY": "nested-secret"}),
            "MINDROOM_KUBERNETES_WORKER_AUTH_SECRET_NAME": "primary-worker-auth",
            "MINDROOM_LOCAL_CLIENT_SECRET": "runtime-client-secret",
            "MINDROOM_SANDBOX_PROXY_TOKEN": "extra-env-token",
            "MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT": "/unsafe/root",
            "MINDROOM_SANDBOX_RUNNER_SUBPROCESS_TIMEOUT_SECONDS": "45",
            "MINDROOM_SHARED_CREDENTIALS_PATH": "/unsafe/shared-credentials",
            "MINDROOM_STORAGE_PATH": "/unsafe/storage",
            "PATH": "/unsafe/bin",
            "VIRTUAL_ENV": "/unsafe/venv",
            "MINDROOM_WORKER_TOOL_VALUE": "visible",
        },
    )

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

    container = apps_api.created_bodies[0]["spec"]["template"]["spec"]["containers"][0]
    env_values = {env["name"]: env.get("value") for env in container["env"]}
    startup_manifest = _load_startup_manifest(backend, worker_key=_TEST_SCOPED_WORKER_KEY_A)
    committed_runtime = deserialize_runtime_paths(startup_manifest["runtime_paths"])
    committed_env = dict(committed_runtime.process_env) | dict(committed_runtime.env_file_values)
    expected_worker_root = f"/app/worker/workers/{worker_dir_name(_TEST_SCOPED_WORKER_KEY_A)}"
    env_names = [env["name"] for env in container["env"]]

    for name in (
        "MINDROOM_KUBERNETES_WORKER_ENV_JSON",
        "MINDROOM_KUBERNETES_WORKER_LABELS_JSON",
        "MINDROOM_KUBERNETES_WORKER_ANNOTATIONS_JSON",
        "MINDROOM_KUBERNETES_WORKER_IMAGE",
        "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME",
        "MINDROOM_KUBERNETES_WORKER_AUTH_SECRET_NAME",
        "MINDROOM_API_KEY",
        "MINDROOM_LOCAL_CLIENT_SECRET",
    ):
        assert name not in env_values
        assert name not in committed_env

    assert env_values["MINDROOM_WORKER_TOOL_VALUE"] == "visible"
    assert committed_runtime.env_value("MINDROOM_WORKER_TOOL_VALUE") == "visible"
    assert env_values["MINDROOM_SANDBOX_RUNNER_SUBPROCESS_TIMEOUT_SECONDS"] == "45"
    assert committed_runtime.env_value("MINDROOM_SANDBOX_RUNNER_SUBPROCESS_TIMEOUT_SECONDS") == "45"
    assert committed_runtime.env_value("MINDROOM_SANDBOX_RUNNER_MODE") == "true"
    assert committed_runtime.env_value("MINDROOM_SANDBOX_RUNNER_EXECUTION_MODE") == "forkserver"
    assert committed_runtime.env_value("MINDROOM_SANDBOX_RUNNER_PORT") == "8766"
    assert env_names.count("MINDROOM_SANDBOX_PROXY_TOKEN") == 1
    assert env_values["MINDROOM_SANDBOX_PROXY_TOKEN"] is None
    assert committed_runtime.env_value("MINDROOM_SANDBOX_PROXY_TOKEN") is None
    assert env_names.count("MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT") == 1
    assert env_values["HOME"] == expected_worker_root
    assert env_values["MINDROOM_CONFIG_PATH"] != "/unsafe/config.yaml"
    assert env_values["MINDROOM_STORAGE_PATH"] == expected_worker_root
    assert env_values["PATH"] != "/unsafe/bin"
    assert env_values["VIRTUAL_ENV"] == f"{expected_worker_root}/venv"
    assert committed_runtime.env_value("MINDROOM_SANDBOX_DEDICATED_WORKER_KEY") == _TEST_SCOPED_WORKER_KEY_A
    assert committed_runtime.env_value("MINDROOM_SANDBOX_DEDICATED_WORKER_ROOT") == expected_worker_root
    assert committed_runtime.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") == (
        f"{expected_worker_root}/.shared_credentials"
    )
    assert committed_runtime.env_value("MINDROOM_CONFIG_PATH") != "/unsafe/config.yaml"
    assert committed_runtime.env_value("MINDROOM_STORAGE_PATH") == expected_worker_root
    assert committed_runtime.env_value("MINDROOM_KUBERNETES_WORKER_STORAGE_SUBPATH_PREFIX") == "workers"


def test_kubernetes_backend_rejects_unknown_worker_keys_for_scoped_mounts() -> None:
    """Malformed worker keys must not fall back to mounting the whole storage root."""
    backend, _apps_api, _core_api = _backend()

    with pytest.raises(WorkerBackendError, match="Unsupported worker key"):
        backend.ensure_worker(WorkerSpec("legacy-worker"), now=10.0)


def test_kubernetes_backend_requires_configured_owner_deployment_to_exist() -> None:
    """Configured owner deployments should fail closed when they cannot be resolved."""
    backend, _apps_api, _core_api = _backend(owner_deployment_name="mindroom-missing")
    assert isinstance(backend._resources.apps_api, _FakeAppsApi)
    backend._resources.apps_api.deployments.pop("mindroom-missing")

    with pytest.raises(WorkerBackendError, match="owner deployment 'mindroom-missing' was not found"):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)


def test_kubernetes_backend_honors_custom_worker_port() -> None:
    """Dedicated workers should wire the configured port through env, service, and probes."""
    backend, apps_api, core_api = _backend(worker_port=9777)

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

    service = core_api.created_bodies[0]
    deployment = apps_api.created_bodies[0]
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_values = {env["name"]: env.get("value") for env in container["env"]}

    assert handle.endpoint == f"http://{handle.worker_id}.chat.svc.cluster.local:9777/api/sandbox-runner/execute"
    assert env_values["MINDROOM_SANDBOX_RUNNER_PORT"] == "9777"
    assert service["spec"]["ports"] == [{"name": "api", "port": 9777, "targetPort": 9777}]
    assert container["ports"] == [{"containerPort": 9777, "name": "api"}]
    assert container["readinessProbe"]["httpGet"]["port"] == "api"
    assert container["livenessProbe"]["httpGet"]["port"] == "api"


def test_kubernetes_backend_mounts_only_scoped_agent_root_for_shared_workers() -> None:
    """Shared-scope dedicated workers should mount only their agent root, not the whole agents tree."""
    backend, apps_api, _core_api = _backend()
    worker_key = "v1:tenant-123:shared:code"

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_worker_root = f"/app/worker/workers/{worker_dir_name(worker_key)}"

    assert mount_paths["/app/worker/agents/code"] == "agents/code"
    assert mount_paths[expected_worker_root] == f"workers/{worker_dir_name(worker_key)}"
    assert "/app/worker/credentials" not in mount_paths
    assert "/app/worker/.shared_credentials" not in mount_paths
    assert "/app/worker/agents" not in mount_paths

    container = deployment["spec"]["template"]["spec"]["containers"][0]
    env_values = {env["name"]: env.get("value") for env in container["env"]}
    assert env_values["MINDROOM_STORAGE_PATH"] == expected_worker_root
    assert env_values["MINDROOM_SHARED_CREDENTIALS_PATH"] == f"{expected_worker_root}/.shared_credentials"


def test_kubernetes_backend_mounts_assigned_knowledge_outside_shared_agent_root_read_only(tmp_path: Path) -> None:
    """Assigned knowledge under shared storage should be visible without widening agent storage."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "knowledge" / "shared-docs"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [shared_docs]
knowledge_bases:
  shared_docs:
    path: ./storage/knowledge/shared-docs
    include_patterns: [docs/**/*.md]
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    knowledge_mount = next(
        mount for mount in volume_mounts if mount["mountPath"] == "/app/worker/knowledge/shared-docs"
    )
    assert knowledge_mount == {
        "name": "worker-storage",
        "mountPath": "/app/worker/knowledge/shared-docs",
        "subPath": "knowledge/shared-docs",
        "readOnly": True,
    }


def test_kubernetes_backend_matches_normalized_agent_name_for_knowledge_mount(tmp_path: Path) -> None:
    """Knowledge selection should match the normalized agent name encoded in a worker key."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "knowledge" / "shared-docs"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  My Agent:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [shared_docs]
knowledge_bases:
  shared_docs:
    path: {knowledge_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)
    worker_key = "v1:tenant-123:shared:My_Agent"

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    assert any(mount.get("subPath") == "knowledge/shared-docs" for mount in volume_mounts)


def test_kubernetes_backend_does_not_duplicate_knowledge_already_visible_in_agent_root(tmp_path: Path) -> None:
    """Knowledge below an existing scoped mount should use that mount without a nested duplicate."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "agents" / "code" / "workspace" / "knowledge"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [workspace_docs, workspace_docs_child]
knowledge_bases:
  workspace_docs:
    path: {knowledge_root}
  workspace_docs_child:
    path: {knowledge_root / "docs"}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = [mount["mountPath"] for mount in volume_mounts]
    assert "/app/worker/agents/code" in mount_paths
    assert "/app/worker/agents/code/workspace/knowledge" not in mount_paths
    assert "/app/worker/agents/code/workspace/knowledge/docs" not in mount_paths


def test_kubernetes_backend_rejects_knowledge_mount_containing_scoped_storage(tmp_path: Path) -> None:
    """A broad knowledge mount must not shadow scoped mounts or expose sibling storage."""
    storage_root = tmp_path / "storage"
    storage_root.mkdir()
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [unsafe_root]
knowledge_bases:
  unsafe_root:
    path: {storage_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

    with pytest.raises(WorkerBackendError, match="knowledge mount overlaps existing mountPath"):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    assert apps_api.created_bodies == []


def test_kubernetes_backend_rejects_overlapping_assigned_knowledge_mounts(tmp_path: Path) -> None:
    """Nested assigned sources should fail closed even if raw config bypassed typed validation."""
    storage_root = tmp_path / "storage"
    parent_root = storage_root / "knowledge" / "project"
    child_root = parent_root / "docs"
    child_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [project, project_docs]
knowledge_bases:
  project:
    path: {parent_root}
  project_docs:
    path: {child_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

    with pytest.raises(WorkerBackendError, match="knowledge mount paths overlap") as exc_info:
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    assert "/app/worker/knowledge/project" in str(exc_info.value)
    assert "/app/worker/knowledge/project/docs" in str(exc_info.value)
    assert apps_api.created_bodies == []


def test_kubernetes_backend_rejects_knowledge_mount_containing_agent_vault_mount(tmp_path: Path) -> None:
    """Knowledge mounts must not contain fixed runner mounts added later in the plan."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "etc"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [unsafe_docs]
knowledge_bases:
  unsafe_docs:
    path: {knowledge_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        storage_mount_path="/",
        agent_vault=_test_agent_vault_config(worker_ca_configmap_name="agent-vault-ca"),
    )

    with pytest.raises(WorkerBackendError, match="knowledge mount overlaps existing mountPath"):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    assert apps_api.created_bodies == []


def test_kubernetes_backend_rejects_knowledge_mount_inside_agent_vault_mount(tmp_path: Path) -> None:
    """Only same-PVC storage ancestors may suppress a redundant knowledge mount."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "agent-vault" / "docs"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [unsafe_docs]
knowledge_bases:
  unsafe_docs:
    path: {knowledge_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        storage_mount_path="/",
        agent_vault=_test_agent_vault_config(),
    )

    with pytest.raises(WorkerBackendError, match="knowledge mount overlaps existing mountPath"):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    assert apps_api.created_bodies == []


def test_kubernetes_backend_does_not_mount_unassigned_knowledge(tmp_path: Path) -> None:
    """Configured knowledge must stay hidden unless assigned to an agent addressed by the worker."""
    storage_root = tmp_path / "storage"
    assigned_root = storage_root / "knowledge" / "assigned"
    unassigned_root = storage_root / "knowledge" / "unassigned"
    assigned_root.mkdir(parents=True)
    unassigned_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [assigned_docs]
knowledge_bases:
  assigned_docs:
    path: {assigned_root}
  unassigned_docs:
    path: {unassigned_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

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

    deployment = apps_api.created_bodies[0]
    mount_paths = {
        mount["mountPath"] for mount in deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    }
    assert "/app/worker/knowledge/assigned" in mount_paths
    assert "/app/worker/knowledge/unassigned" not in mount_paths


def test_kubernetes_backend_ignores_assigned_knowledge_outside_worker_storage(tmp_path: Path) -> None:
    """Unsupported sources outside the shared PVC should not change existing scoped mounts."""
    storage_root = tmp_path / "storage"
    external_root = tmp_path / "external-docs"
    external_root.mkdir()
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [external_docs]
knowledge_bases:
  external_docs:
    path: {external_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"] for mount in volume_mounts}
    expected_worker_root = f"/app/worker/workers/{worker_dir_name(_TEST_SCOPED_WORKER_KEY_A)}"
    assert mount_paths == {"/app/worker/agents/code", expected_worker_root, "/app/config.yaml"}


def test_kubernetes_backend_user_worker_mounts_knowledge_for_all_addressed_agents(tmp_path: Path) -> None:
    """One user-scoped worker should see assignments from user-scoped agents, not other scopes."""
    storage_root = tmp_path / "storage"
    for dirname in ("alpha", "beta", "gamma"):
        (storage_root / "knowledge" / dirname).mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    worker_scope: user
    knowledge_bases: [alpha_docs]
  beta:
    display_name: Beta
    role: Beta test
    model: default
    worker_scope: user
    knowledge_bases: [beta_docs]
  gamma:
    display_name: Gamma
    role: Gamma test
    model: default
    worker_scope: shared
    knowledge_bases: [gamma_docs]
knowledge_bases:
  alpha_docs:
    path: {storage_root / "knowledge" / "alpha"}
  beta_docs:
    path: {storage_root / "knowledge" / "beta"}
  gamma_docs:
    path: {storage_root / "knowledge" / "gamma"}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)
    worker_key = "v1:tenant-123:user:@alice:example.org"

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

    deployment = apps_api.created_bodies[0]
    mount_paths = {
        mount["mountPath"] for mount in deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    }
    assert "/app/worker/knowledge/alpha" in mount_paths
    assert "/app/worker/knowledge/beta" in mount_paths
    assert "/app/worker/knowledge/gamma" not in mount_paths


def test_kubernetes_backend_recreates_worker_when_knowledge_mounts_change(tmp_path: Path) -> None:
    """Knowledge mount changes must alter the pod-template hash and recreate a stale Deployment."""
    storage_root = tmp_path / "storage"
    knowledge_root = storage_root / "knowledge" / "shared-docs"
    knowledge_root.mkdir(parents=True)
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
knowledge_bases:
  shared_docs:
    path: {knowledge_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path, storage_path=storage_root)
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    initial_hash = apps_api.created_bodies[0]["metadata"]["annotations"][_ANNOTATION_TEMPLATE_HASH]

    config_path.write_text(
        f"""
agents:
  code:
    display_name: Code
    role: Code test
    model: default
    worker_scope: shared
    knowledge_bases: [shared_docs]
knowledge_bases:
  shared_docs:
    path: {knowledge_root}
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    updated_backend, _updated_apps_api, _updated_core_api = _backend(runtime_paths=runtime_paths)
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError

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

    recreated = apps_api.created_bodies[-1]
    updated_hash = recreated["metadata"]["annotations"][_ANNOTATION_TEMPLATE_HASH]
    assert apps_api.deleted_names == [recreated["metadata"]["name"]]
    assert updated_hash != initial_hash
    volume_mounts = recreated["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    assert any(mount["subPath"] == "knowledge/shared-docs" for mount in volume_mounts)


def test_kubernetes_backend_uses_custom_worker_prefix_for_storage_path() -> None:
    """Custom worker prefixes should only affect the dedicated worker storage root."""
    backend, apps_api, _core_api = _backend(storage_subpath_prefix="sandbox-workers")
    worker_key = "v1:tenant-123:shared:code"

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

    deployment = apps_api.created_bodies[0]
    env_values = {
        env["name"]: env.get("value") for env in deployment["spec"]["template"]["spec"]["containers"][0]["env"]
    }
    expected_worker_root = f"/app/worker/sandbox-workers/{worker_dir_name(worker_key)}"

    assert env_values["MINDROOM_STORAGE_PATH"] == expected_worker_root


def test_kubernetes_backend_mounts_broad_agents_tree_for_user_scope() -> None:
    """User-scope workers should see shared agents plus their own private-instance namespace."""
    backend, apps_api, _core_api = _backend()
    worker_key = "v1:tenant-123:user:@alice:example.org"

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_worker_root = f"/app/worker/workers/{worker_dir_name(worker_key)}"
    expected_private_root = f"/app/worker/private_instances/{worker_dir_name(worker_key)}"

    assert mount_paths["/app/worker/agents"] == "agents"
    assert mount_paths[expected_private_root] == f"private_instances/{worker_dir_name(worker_key)}"
    assert mount_paths[expected_worker_root] == f"workers/{worker_dir_name(worker_key)}"
    assert "/app/worker/credentials" not in mount_paths
    assert "/app/worker/.shared_credentials" not in mount_paths


def test_kubernetes_backend_user_agent_mounts_require_explicit_private_visibility(tmp_path: Path) -> None:
    """User-agent mounts should fail closed without explicit private visibility."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )
    backend, _apps_api, _core_api = _backend(runtime_paths=runtime_paths)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )

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


def test_kubernetes_backend_rejects_private_user_agent_worker_without_target_visibility(tmp_path: Path) -> None:
    """Private user-agent workers must fail closed until the targeted private agent is explicitly visible."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        """
agents:
  alpha:
    display_name: Alpha
    role: Alpha test
    model: default
    private:
      per: user_agent
models:
  default:
    provider: openai
    id: gpt-5.4
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
    )
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)
    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 apps_api.created_bodies == []


def test_kubernetes_backend_user_agent_mounts_private_root_from_worker_spec() -> None:
    """User-agent workers should mount their private root from the explicit worker spec visibility."""
    backend, apps_api, _core_api = _backend()
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_private_root = str(
        _private_instance_state_root_path(
            Path("/app/worker"),
            worker_key=worker_key,
            agent_name="mind",
        ),
    )
    expected_private_subpath = f"private_instances/{worker_dir_name(worker_key)}/mind"

    assert mount_paths[expected_private_root] == expected_private_subpath
    assert "/app/worker/agents/mind" not in mount_paths
    assert f"/app/worker/private_instances/{worker_dir_name(worker_key)}" not in mount_paths


def test_kubernetes_backend_recreates_user_agent_deployment_when_private_visibility_changes() -> None:
    """Changing private visibility should recreate the Deployment instead of relying on patch semantics."""
    backend, apps_api, _core_api = _backend()
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )

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

    assert len(apps_api.created_bodies) == 2
    recreated_worker_id = apps_api.created_bodies[-1]["metadata"]["name"]
    assert apps_api.deleted_names == [recreated_worker_id]
    recreated = apps_api.created_bodies[-1]
    volume_mounts = recreated["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_private_root = str(
        _private_instance_state_root_path(
            Path("/app/worker"),
            worker_key=worker_key,
            agent_name="mind",
        ),
    )
    expected_private_subpath = f"private_instances/{worker_dir_name(worker_key)}/mind"

    assert mount_paths[expected_private_root] == expected_private_subpath
    assert "/app/worker/agents/mind" not in mount_paths


def test_kubernetes_backend_waits_for_deployment_deletion_before_recreate() -> None:
    """Template-drift replacement should wait for actual deletion instead of patching a terminating Deployment."""
    backend, apps_api, _core_api = _backend()
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )

    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset()), now=10.0)
    worker_id = apps_api.created_bodies[0]["metadata"]["name"]
    apps_api.delete_read_lag_by_name[worker_id] = 1

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

    assert apps_api.deleted_names == [worker_id]
    assert len(apps_api.created_bodies) == 2
    recreated = apps_api.created_bodies[-1]
    volume_mounts = recreated["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_private_root = str(
        _private_instance_state_root_path(
            Path("/app/worker"),
            worker_key=worker_key,
            agent_name="mind",
        ),
    )

    assert mount_paths[expected_private_root] == f"private_instances/{worker_dir_name(worker_key)}/mind"


def test_kubernetes_backend_mounts_only_scoped_agent_root_for_unscoped_workers() -> None:
    """Unscoped dedicated workers should mount only the addressed agent root."""
    backend, apps_api, _core_api = _backend()
    worker_key = resolve_unscoped_worker_key(agent_name="general")

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

    deployment = apps_api.created_bodies[0]
    volume_mounts = deployment["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    mount_paths = {mount["mountPath"]: mount.get("subPath") for mount in volume_mounts}
    expected_worker_root = f"/app/worker/workers/{worker_dir_name(worker_key)}"

    assert mount_paths["/app/worker/agents/general"] == "agents/general"
    assert mount_paths[expected_worker_root] == f"workers/{worker_dir_name(worker_key)}"
    assert "/app/worker/agents" not in mount_paths
    assert "/app/worker/credentials" not in mount_paths
    assert "/app/worker/.shared_credentials" not in mount_paths


def test_kubernetes_backend_seeds_ui_shared_credentials_for_unscoped_workers(monkeypatch: pytest.MonkeyPatch) -> None:
    """Unscoped dedicated workers should mirror shared UI credentials into their shared layer."""
    backend, _apps_api, _core_api = _backend()
    sync_calls: list[tuple[str, frozenset[str] | None]] = []

    def _record_sync(
        worker_key: str,
        *,
        allowed_services: frozenset[str] | None = None,
        credentials_manager: object | None = None,
    ) -> None:
        del credentials_manager
        sync_calls.append((worker_key, allowed_services))

    monkeypatch.setattr(
        "mindroom.workers.backends.kubernetes.sync_shared_credentials_to_worker",
        _record_sync,
    )

    backend.ensure_worker(WorkerSpec("v1:tenant-123:unscoped:general"), now=10.0)

    assert sync_calls == [("v1:tenant-123:unscoped:general", DEFAULT_WORKER_GRANTABLE_CREDENTIALS)]


def test_kubernetes_backend_mirrors_shared_credentials_for_scoped_workers(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Scoped dedicated workers should use the same allowlisted shared-credential sync path."""
    backend, _apps_api, _core_api = _backend()
    sync_calls: list[tuple[str, frozenset[str] | None]] = []

    def _record_sync(
        worker_key: str,
        *,
        allowed_services: frozenset[str] | None = None,
        credentials_manager: object | None = None,
    ) -> None:
        del credentials_manager
        sync_calls.append((worker_key, allowed_services))

    monkeypatch.setattr(
        "mindroom.workers.backends.kubernetes.sync_shared_credentials_to_worker",
        _record_sync,
    )

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

    assert sync_calls == [("v1:tenant-123:user:@alice:example.org", DEFAULT_WORKER_GRANTABLE_CREDENTIALS)]


def test_kubernetes_backend_uses_configured_worker_grantable_credentials(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedicated workers should use the config-authored worker credential allowlist."""
    backend, _apps_api, _core_api = _backend(
        worker_grantable_credentials=frozenset({"openai", "github_private"}),
    )
    sync_calls: list[frozenset[str] | None] = []

    def _record_sync(
        worker_key: str,
        *,
        allowed_services: frozenset[str] | None = None,
        credentials_manager: object | None = None,
    ) -> None:
        del worker_key, credentials_manager
        sync_calls.append(allowed_services)

    monkeypatch.setattr(
        "mindroom.workers.backends.kubernetes.sync_shared_credentials_to_worker",
        _record_sync,
    )

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

    assert sync_calls == [frozenset({"openai", "github_private"})]


def test_kubernetes_backend_uses_empty_worker_grantable_credentials_allowlist(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedicated workers should pass through an explicit deny-all worker credential allowlist."""
    backend, _apps_api, _core_api = _backend(worker_grantable_credentials=frozenset())
    sync_calls: list[frozenset[str] | None] = []

    def _record_sync(
        worker_key: str,
        *,
        allowed_services: frozenset[str] | None = None,
        credentials_manager: object | None = None,
    ) -> None:
        del worker_key, credentials_manager
        sync_calls.append(allowed_services)

    monkeypatch.setattr(
        "mindroom.workers.backends.kubernetes.sync_shared_credentials_to_worker",
        _record_sync,
    )

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

    assert sync_calls == [frozenset()]


def test_kubernetes_backend_cleanup_scales_idle_workers_to_zero() -> None:
    """Idle cleanup should scale dedicated workers to zero while keeping their metadata."""
    backend, apps_api, core_api = _backend(idle_timeout_seconds=5.0)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    deployment = apps_api.deployments[handle.worker_id]
    deployment.metadata.annotations["mindroom.ai/last-used-at"] = "0.0"
    deployment.metadata.annotations["mindroom.ai/worker-status"] = "ready"

    cleaned = backend.cleanup_idle_workers(now=10.0)

    assert len(cleaned) == 1
    assert cleaned[0].worker_key == _TEST_SCOPED_WORKER_KEY_A
    assert cleaned[0].status == "idle"
    assert deployment.spec.replicas == 0
    assert handle.worker_id not in core_api.services
    assert handle.worker_id not in core_api.secrets


def test_kubernetes_backend_cleanup_is_idempotent_for_already_idle_workers() -> None:
    """Cleanup should not report or patch workers that are already scaled to zero."""
    backend, apps_api, _core_api = _backend(idle_timeout_seconds=5.0)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    deployment = apps_api.deployments[handle.worker_id]
    deployment.metadata.annotations["mindroom.ai/last-used-at"] = "0.0"
    deployment.metadata.annotations["mindroom.ai/worker-status"] = "ready"

    first_cleaned = backend.cleanup_idle_workers(now=10.0)
    patch_count_after_first_cleanup = len(apps_api.patched_bodies)
    second_cleaned = backend.cleanup_idle_workers(now=11.0)

    assert [worker.worker_key for worker in first_cleaned] == [_TEST_SCOPED_WORKER_KEY_A]
    assert second_cleaned == []
    assert len(apps_api.patched_bodies) == patch_count_after_first_cleanup


def test_kubernetes_backend_cleanup_idle_deletes_service_but_keeps_deployment() -> None:
    """Idle cleanup should scale down the worker and release its Service."""
    backend, apps_api, core_api = _backend()
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)

    cleaned = backend.cleanup_idle_workers(now=80.0)

    assert [worker.worker_key for worker in cleaned] == [_TEST_SCOPED_WORKER_KEY_A]
    assert cleaned[0].status == "idle"
    assert handle.worker_id in apps_api.deployments
    assert apps_api.deployments[handle.worker_id].spec.replicas == 0
    assert handle.worker_id not in core_api.services
    assert handle.worker_id not in core_api.secrets


def _wire_fake_apis(backend: KubernetesWorkerBackend, apps_api: _FakeAppsApi, core_api: _FakeCoreApi) -> None:
    backend._resources.apps_api = apps_api
    backend._resources.core_api = core_api
    backend._resources.api_exception_cls = _FakeApiError


def _kubernetes_api_call_counts(apps_api: _FakeAppsApi, core_api: _FakeCoreApi) -> tuple[int, ...]:
    return (
        len(apps_api.read_names),
        len(apps_api.created_bodies),
        len(apps_api.patched_bodies),
        len(apps_api.deleted_names),
        len(core_api.read_service_names),
        len(core_api.created_bodies),
        len(core_api.read_secret_names),
        len(core_api.created_secret_bodies),
        len(core_api.patched_secret_bodies),
        len(core_api.deleted_secret_names),
    )


def test_kubernetes_backend_reuses_recent_ready_worker_with_one_metadata_patch() -> None:
    """A recently validated ready worker should skip cluster reconciliation."""
    backend, apps_api, core_api = _backend()
    first = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    calls_after_first_ensure = _kubernetes_api_call_counts(apps_api, core_api)

    second = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=20.0)

    assert second.worker_id == first.worker_id
    assert second.last_used_at == 20.0
    calls_after_second_ensure = _kubernetes_api_call_counts(apps_api, core_api)
    assert tuple(
        after - before for before, after in zip(calls_after_first_ensure, calls_after_second_ensure, strict=True)
    ) == (
        0,
        0,
        1,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
    )
    assert apps_api.deployments[first.worker_id].metadata.annotations["mindroom.ai/last-used-at"] == "20.0"


def test_kubernetes_backend_reconciles_when_cached_worker_disappears() -> None:
    """A missing cached Deployment should fall back to full reconciliation."""
    backend, apps_api, _core_api = _backend()
    first = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    apps_api.deployments.pop(first.worker_id)

    recreated = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=20.0)

    assert recreated.status == "ready"
    assert len(apps_api.created_bodies) == 2


def test_kubernetes_backend_periodically_revalidates_cached_ready_worker() -> None:
    """The ready cache should bound how long external cluster drift can stay hidden."""
    backend, apps_api, core_api = _backend(idle_timeout_seconds=600.0)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    calls_after_first_ensure = _kubernetes_api_call_counts(apps_api, core_api)

    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=309.0)
    calls_before_revalidation = _kubernetes_api_call_counts(apps_api, core_api)
    assert calls_before_revalidation[0] == calls_after_first_ensure[0]
    assert calls_before_revalidation[2] == calls_after_first_ensure[2] + 1
    assert calls_before_revalidation[4:] == calls_after_first_ensure[4:]

    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=310.0)
    calls_after_revalidation = _kubernetes_api_call_counts(apps_api, core_api)
    assert calls_after_revalidation[0] > calls_before_revalidation[0]
    assert calls_after_revalidation[4] > calls_before_revalidation[4]
    assert calls_after_revalidation[6] > calls_before_revalidation[6]


def test_kubernetes_backend_cached_touch_uses_one_metadata_patch() -> None:
    """A cached touch should persist usage without reading or reconciling cluster objects."""
    backend, apps_api, _core_api = _backend()
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    patch_count_after_ensure = len(apps_api.patched_bodies)
    read_count_after_ensure = len(apps_api.read_names)

    first_touch = backend.touch_worker(_TEST_SCOPED_WORKER_KEY_A, now=11.0)
    second_touch = backend.touch_worker(_TEST_SCOPED_WORKER_KEY_A, now=12.0)

    assert first_touch is not None
    assert second_touch is not None
    assert second_touch.last_used_at == 12.0
    assert len(apps_api.patched_bodies) == patch_count_after_ensure + 2
    assert len(apps_api.read_names) == read_count_after_ensure
    assert backend.list_workers(now=12.0)[0].last_used_at == 12.0
    assert apps_api.deployments[handle.worker_id].metadata.annotations["mindroom.ai/last-used-at"] == "12.0"
    assert apps_api.deployments[handle.worker_id].metadata.annotations[ANNOTATION_WORKER_KEY] == (
        _TEST_SCOPED_WORKER_KEY_A
    )


def test_kubernetes_backend_failure_invalidates_ready_cache() -> None:
    """A failed worker must take the full restart path on its next ensure."""
    backend, apps_api, core_api = _backend()
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    backend.record_failure(_TEST_SCOPED_WORKER_KEY_A, "worker failed", now=11.0)
    calls_after_failure = _kubernetes_api_call_counts(apps_api, core_api)

    restarted = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=12.0)

    assert restarted.status == "ready"
    assert restarted.startup_count == 2
    assert _kubernetes_api_call_counts(apps_api, core_api) != calls_after_failure


def test_kubernetes_backend_reconciles_drifted_idle_worker_template(tmp_path: Path) -> None:
    """Reconciliation should recreate scaled-down workers whose pod template drifted from current config."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    backend.cleanup_idle_workers(now=80.0)

    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)

    reconciled = updated_backend.maintain_workers(now=90.0).reconciled

    assert [worker.worker_key for worker in reconciled] == [_TEST_SCOPED_WORKER_KEY_A]
    assert reconciled[0].status == "idle"
    assert apps_api.deleted_names == [handle.worker_id]
    assert len(apps_api.created_bodies) == 2
    recreated = apps_api.created_bodies[1]
    assert recreated["spec"]["replicas"] == 0
    container = recreated["spec"]["template"]["spec"]["containers"][0]
    assert container["resources"]["limits"] == {"memory": "2Gi", "cpu": "1"}
    assert recreated["metadata"]["annotations"]["mindroom.ai/created-at"] == "0.0"


def test_kubernetes_backend_maintenance_lists_deployments_once(tmp_path: Path) -> None:
    """Cleanup and reconciliation should share one lightweight Kubernetes list response."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)
    list_count_before = len(apps_api.list_label_selectors)

    maintenance = updated_backend.maintain_workers(now=80.0)

    assert len(apps_api.list_label_selectors) == list_count_before + 1
    assert apps_api.raw_list_count == 1
    assert [worker.worker_key for worker in maintenance.cleaned] == [_TEST_SCOPED_WORKER_KEY_A]
    assert [worker.worker_key for worker in maintenance.reconciled] == [_TEST_SCOPED_WORKER_KEY_A]


def test_kubernetes_backend_reconcile_defers_running_workers(tmp_path: Path) -> None:
    """Reconciliation should leave running workers to the ensure-time template-hash check."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)

    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)

    reconciled = updated_backend.maintain_workers(now=10.0).reconciled

    assert reconciled == ()
    assert apps_api.deleted_names == []
    assert len(apps_api.created_bodies) == 1


def test_kubernetes_backend_reconcile_leaves_unchanged_templates_alone(tmp_path: Path) -> None:
    """Reconciliation should not touch scaled-down workers whose template still matches current config."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    backend.cleanup_idle_workers(now=80.0)
    patch_count_after_cleanup = len(apps_api.patched_bodies)

    unchanged_backend, _, _ = _backend(runtime_paths=runtime_paths)
    _wire_fake_apis(unchanged_backend, apps_api, core_api)

    reconciled = unchanged_backend.maintain_workers(now=90.0).reconciled

    assert reconciled == ()
    assert apps_api.deleted_names == []
    assert len(apps_api.created_bodies) == 1
    assert len(apps_api.patched_bodies) == patch_count_after_cleanup


def test_kubernetes_backend_reconcile_disabled_by_config(tmp_path: Path) -> None:
    """Disabling pod-template reconciliation should keep drifted scaled-down workers untouched."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    backend.cleanup_idle_workers(now=80.0)

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        resource_limits={"memory": "2Gi", "cpu": "1"},
        reconcile_pod_templates=False,
    )
    _wire_fake_apis(updated_backend, apps_api, core_api)

    reconciled = updated_backend.maintain_workers(now=90.0).reconciled

    assert reconciled == ()
    assert apps_api.deleted_names == []
    assert len(apps_api.created_bodies) == 1


def test_kubernetes_backend_reconcile_uses_persisted_private_visibility(tmp_path: Path) -> None:
    """Reconciliation should rebuild user-agent worker templates from persisted private visibility."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )
    backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"mind"})), now=0.0)
    backend.cleanup_idle_workers(now=80.0)

    unchanged_backend, _, _ = _backend(runtime_paths=runtime_paths)
    _wire_fake_apis(unchanged_backend, apps_api, core_api)
    assert unchanged_backend.maintain_workers(now=90.0).reconciled == ()

    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)

    reconciled = updated_backend.maintain_workers(now=100.0).reconciled

    assert [worker.worker_key for worker in reconciled] == [worker_key]
    recreated = apps_api.created_bodies[-1]
    assert recreated["metadata"]["annotations"][_ANNOTATION_PRIVATE_AGENT_NAMES] == '["mind"]'
    mount_paths = {
        mount["mountPath"] for mount in recreated["spec"]["template"]["spec"]["containers"][0]["volumeMounts"]
    }
    expected_private_root = str(
        _private_instance_state_root_path(
            Path("/app/worker"),
            worker_key=worker_key,
            agent_name="mind",
        ),
    )
    assert expected_private_root in mount_paths


def test_kubernetes_backend_reconcile_revalidates_live_state_before_recreating(tmp_path: Path) -> None:
    """A worker provisioned between the list snapshot and the lock must not be scaled back down."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)
    backend.cleanup_idle_workers(now=80.0)
    stale_deployment = deepcopy(apps_api.deployments[handle.worker_id])

    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)
    updated_backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=85.0)

    def _stale_snapshot() -> list[object]:
        return [stale_deployment]

    updated_backend._resources.list_deployments = _stale_snapshot

    reconciled = updated_backend.maintain_workers(now=90.0).reconciled

    assert reconciled == ()
    live_deployment = apps_api.deployments[handle.worker_id]
    assert int(live_deployment.spec.replicas) == 1
    assert live_deployment.metadata.annotations[ANNOTATION_WORKER_KEY] == _TEST_SCOPED_WORKER_KEY_A
    assert (
        live_deployment.metadata.annotations[_ANNOTATION_TEMPLATE_HASH]
        != stale_deployment.metadata.annotations[_ANNOTATION_TEMPLATE_HASH]
    )


def test_kubernetes_backend_reconcile_defers_user_agent_workers_without_persisted_visibility(
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """User-agent workers without persisted visibility defer to ensure-time recreation without warning each pass."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths)
    worker_key = resolve_worker_key(
        "user_agent",
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="mind",
            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="mind",
    )
    handle = backend.ensure_worker(WorkerSpec(worker_key, private_agent_names=frozenset({"mind"})), now=0.0)
    backend.cleanup_idle_workers(now=80.0)
    del apps_api.deployments[handle.worker_id].metadata.annotations[_ANNOTATION_PRIVATE_AGENT_NAMES]

    updated_backend, _, _ = _backend(runtime_paths=runtime_paths, resource_limits={"memory": "2Gi", "cpu": "1"})
    _wire_fake_apis(updated_backend, apps_api, core_api)

    with caplog.at_level("WARNING", logger=kubernetes_backend_module.__name__):
        reconciled = updated_backend.maintain_workers(now=90.0).reconciled

    assert reconciled == ()
    assert apps_api.deleted_names == []
    assert len(apps_api.created_bodies) == 1
    assert caplog.records == []


def test_kubernetes_backend_config_reads_reconcile_pod_templates_from_env(tmp_path: Path) -> None:
    """Pod-template reconciliation defaults on and is disabled through runtime env."""
    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.6\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    base_env = (
        "MINDROOM_WORKER_BACKEND=kubernetes\n"
        "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
        "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
    )
    (config_dir / ".env").write_text(base_env, encoding="utf-8")

    default_config = KubernetesWorkerBackendConfig.from_runtime(resolve_primary_runtime_paths(config_path=config_path))

    assert default_config.reconcile_pod_templates is True

    (config_dir / ".env").write_text(
        base_env + "MINDROOM_KUBERNETES_WORKER_RECONCILE_POD_TEMPLATES=false\n",
        encoding="utf-8",
    )

    disabled_config = KubernetesWorkerBackendConfig.from_runtime(
        resolve_primary_runtime_paths(config_path=config_path),
    )

    assert disabled_config.reconcile_pod_templates is False


def test_kubernetes_backend_list_workers_is_scoped_to_backend_labels() -> None:
    """Worker discovery should stay confined to this backend's label set within a shared namespace."""
    backend, apps_api, _core_api = _backend()
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=0.0)

    unrelated_body = deepcopy(apps_api.created_bodies[0])
    unrelated_name = "mindroom-worker-unrelated"
    unrelated_body["metadata"]["name"] = unrelated_name
    unrelated_body["metadata"]["annotations"]["mindroom.ai/worker-key"] = _TEST_SCOPED_WORKER_KEY_B
    unrelated_body["metadata"]["labels"]["mindroom.ai/tenant"] = "other"
    unrelated_body["metadata"]["labels"]["mindroom.ai/worker-id"] = unrelated_name
    unrelated_body["spec"]["selector"]["matchLabels"]["mindroom.ai/tenant"] = "other"
    unrelated_body["spec"]["selector"]["matchLabels"]["mindroom.ai/worker-id"] = unrelated_name
    unrelated_body["spec"]["template"]["metadata"]["labels"]["mindroom.ai/tenant"] = "other"
    unrelated_body["spec"]["template"]["metadata"]["labels"]["mindroom.ai/worker-id"] = unrelated_name
    unrelated_body["spec"]["replicas"] = 1
    unrelated = _to_namespace(unrelated_body)
    unrelated.metadata.generation = 1
    unrelated.status = SimpleNamespace(ready_replicas=1, observed_generation=1)
    apps_api.deployments[unrelated_name] = unrelated

    workers = backend.list_workers(now=10.0)

    assert [worker.worker_key for worker in workers] == [handle.worker_key]
    assert apps_api.list_label_selectors[-1] == (
        "app.kubernetes.io/managed-by=mindroom,"
        "app.kubernetes.io/name=mindroom-worker,"
        "mindroom.ai/component=worker,"
        "mindroom.ai/tenant=test"
    )


def test_kubernetes_backend_touch_only_patches_deployment_metadata() -> None:
    """Refreshing worker usage must not mutate the pod template and trigger a rollout."""
    backend, apps_api, _core_api = _backend()
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    touched = backend.touch_worker(_TEST_SCOPED_WORKER_KEY_A, now=25.0)

    assert touched is not None
    patch_name, patch_body = apps_api.patched_bodies[-1]
    assert patch_name == handle.worker_id
    assert patch_body["metadata"]["annotations"]["mindroom.ai/last-used-at"] == "25.0"
    assert "template" not in patch_body.get("spec", {})


def test_kubernetes_backend_touch_revives_idle_worker_and_clears_stale_failure_reason() -> None:
    """A touch must revive an idle worker and drop a lingering failure reason."""
    backend, apps_api, _core_api = _backend()
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    deployment = apps_api.deployments[handle.worker_id]
    deployment.metadata.annotations["mindroom.ai/worker-status"] = "idle"
    deployment.metadata.annotations["mindroom.ai/failure-reason"] = "boom"

    touched = backend.touch_worker(_TEST_SCOPED_WORKER_KEY_A, now=25.0)

    assert touched is not None
    assert touched.status == "ready"
    assert touched.failure_reason is None


def test_kubernetes_backend_pins_workers_to_control_plane_node(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedicated workers should co-locate with the control-plane pod when using a shared RWO PVC."""
    backend, apps_api, core_api = _backend(colocate_with_control_plane_node=True)
    core_api.pods["mindroom-control-plane"] = SimpleNamespace(
        metadata=SimpleNamespace(name="mindroom-control-plane"),
        spec=SimpleNamespace(node_name="gke-chat-node-1"),
    )
    monkeypatch.setenv("HOSTNAME", "mindroom-control-plane")

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

    deployment = apps_api.created_bodies[0]
    assert deployment["spec"]["template"]["spec"]["nodeName"] == "gke-chat-node-1"


def test_kubernetes_backend_uses_explicit_worker_node_name_when_configured() -> None:
    """Dedicated workers should honor an explicit node pin without querying the control-plane pod."""
    backend, apps_api, _core_api = _backend(node_name="gke-chat-node-2")

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

    deployment = apps_api.created_bodies[0]
    assert deployment["spec"]["template"]["spec"]["nodeName"] == "gke-chat-node-2"


def test_kubernetes_backend_does_not_pin_workers_when_colocation_disabled() -> None:
    """RWX-capable deployments should be able to omit node pinning entirely."""
    backend, apps_api, _core_api = _backend()

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

    deployment = apps_api.created_bodies[0]
    assert "nodeName" not in deployment["spec"]["template"]["spec"]


def test_kubernetes_backend_records_failed_startup_state() -> None:
    """Workers that never become ready should surface as failed instead of starting forever."""
    backend, apps_api, _core_api = _backend()
    error_message = "worker never became ready"

    def _boom(
        _self: object,
        _deployment_name: str,
        *,
        timeout_seconds: float,
        deployment_ready_fn: object,
        on_poll_tick: object | None = None,
    ) -> object:
        del timeout_seconds, deployment_ready_fn, on_poll_tick
        raise WorkerBackendError(error_message)

    backend._resources.wait_for_ready = MethodType(_boom, backend._resources)

    with pytest.raises(WorkerBackendError, match=error_message):
        backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    worker_id = next(iter(apps_api.deployments))
    deployment = apps_api.deployments[worker_id]
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] == "failed"
    assert deployment.metadata.annotations["mindroom.ai/failure-reason"] == error_message
    assert deployment.metadata.annotations["mindroom.ai/failure-count"] == "1"
    assert deployment.spec.replicas == 0

    handle = backend.list_workers(now=11.0)[0]
    assert handle.status == "failed"
    assert handle.failure_reason == error_message
    assert worker_id not in _core_api.services
    assert worker_id not in _core_api.secrets


def test_kubernetes_backend_progress_respects_real_elapsed_time(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Cold-start progress should fire on the 1.5s grace deadline even with 1.0s polling."""
    backend, _apps_api, _core_api = _backend()
    events: list[WorkerReadyProgress] = []
    cold_start_seen = threading.Event()
    ensure_worker_returned = threading.Event()
    first_poll_seen = threading.Event()
    clock = _ControlledMonotonicClock()
    handle: list[object] = []
    errors: list[BaseException] = []

    def sink(progress: WorkerReadyProgress) -> None:
        events.append(progress)
        if progress.phase == "cold_start":
            cold_start_seen.set()

    monkeypatch.setattr(
        kubernetes_backend_module,
        "time",
        SimpleNamespace(monotonic=clock.monotonic, time=time.time),
    )
    monkeypatch.setattr(
        kubernetes_backend_module,
        "threading",
        SimpleNamespace(
            Condition=lambda: _ControlledCondition(clock),
            Thread=threading.Thread,
            Lock=threading.Lock,
        ),
    )
    _install_real_elapsed_wait_for_ready(
        backend,
        ready_after_seconds=1.7,
        poll_interval_seconds=kubernetes_resources_module._READY_POLL_INTERVAL_SECONDS,
        monotonic=clock.monotonic,
        sleep=clock.sleep,
        on_iteration=lambda _elapsed_seconds: first_poll_seen.set(),
    )

    def ensure_worker() -> None:
        try:
            handle.append(
                backend.ensure_worker(
                    WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
                    now=10.0,
                    progress_sink=sink,
                ),
            )
        except BaseException as exc:  # pragma: no cover - raised explicitly below
            errors.append(exc)
        finally:
            ensure_worker_returned.set()

    worker_thread = threading.Thread(target=ensure_worker)
    worker_thread.start()

    assert first_poll_seen.wait(timeout=1.0)
    clock.advance_to(1.0)
    assert not cold_start_seen.wait(timeout=0.1)
    assert not ensure_worker_returned.is_set()

    clock.advance_to(1.5)
    assert cold_start_seen.wait(timeout=1.0)
    assert not ensure_worker_returned.is_set()

    clock.advance_to(1.7)
    assert not ensure_worker_returned.is_set()

    clock.advance_to(2.0)
    worker_thread.join(timeout=1.0)
    assert ensure_worker_returned.is_set()

    if errors:
        raise errors[0]

    assert len(handle) == 1
    assert [event.phase for event in events] == ["cold_start", "ready"]
    assert 1.5 <= events[0].elapsed_seconds <= 1.7
    assert events[1].elapsed_seconds >= 2.0

    assert handle[0].worker_key == _TEST_SCOPED_WORKER_KEY_A


def test_kubernetes_backend_skips_progress_for_warm_worker() -> None:
    """Warm ready workers should stay silent even when a sink is provided."""
    backend, _apps_api, _core_api = _backend()
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    events: list[WorkerReadyProgress] = []

    backend.ensure_worker(
        WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
        now=11.0,
        progress_sink=events.append,
    )

    assert events == []


def test_kubernetes_backend_reports_progress_for_recreated_ready_deployment(tmp_path: Path) -> None:
    """Recreating a ready deployment should emit cold-start progress for the real startup lifecycle."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    initial_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot["search"] = {
        "config_fields": ["engine"],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    }
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=initial_snapshot,
    )
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=updated_snapshot,
    )
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError
    _install_real_elapsed_wait_for_ready(updated_backend, ready_after_seconds=1.6)

    events: list[WorkerReadyProgress] = []
    updated_backend.ensure_worker(
        WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
        now=11.0,
        progress_sink=events.append,
    )

    assert [event.phase for event in events] == ["cold_start", "ready"]


def test_kubernetes_backend_recreated_ready_deployment_refreshes_startup_metadata(tmp_path: Path) -> None:
    """Recreating a ready deployment should refresh startup metadata instead of reusing stale values."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    initial_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot["search"] = {
        "config_fields": ["engine"],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    }
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=initial_snapshot,
    )
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=updated_snapshot,
    )
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError
    _install_real_elapsed_wait_for_ready(updated_backend, ready_after_seconds=1.6)

    handle = updated_backend.ensure_worker(
        WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
        now=11.0,
    )

    assert handle.last_started_at == 11.0
    assert handle.startup_count == 2


def test_kubernetes_backend_recreate_metadata_patch_failure_is_normalized(tmp_path: Path) -> None:
    """Recreate metadata patch failures should still record failed startup state and raise WorkerBackendError."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    initial_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot["search"] = {
        "config_fields": ["engine"],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    }
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=initial_snapshot,
    )
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=updated_snapshot,
    )
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError

    original_patch_deployment = updated_backend._resources.patch_deployment
    first_patch_attempt = True

    def patch_deployment_with_failure(
        deployment_name: str,
        *,
        replicas: int | None = None,
        annotations: dict[str, str] | None = None,
    ) -> None:
        nonlocal first_patch_attempt
        if first_patch_attempt:
            first_patch_attempt = False
            msg = "refresh metadata failed"
            raise RuntimeError(msg)
        original_patch_deployment(
            deployment_name,
            replicas=replicas,
            annotations=annotations,
        )

    updated_backend._resources.patch_deployment = patch_deployment_with_failure

    events: list[WorkerReadyProgress] = []
    with pytest.raises(WorkerBackendError, match="refresh metadata failed"):
        updated_backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=11.0,
            progress_sink=events.append,
        )

    worker_id = next(iter(apps_api.deployments))
    deployment = apps_api.deployments[worker_id]
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] == "failed"
    assert deployment.metadata.annotations["mindroom.ai/failure-reason"] == "refresh metadata failed"
    assert deployment.metadata.annotations["mindroom.ai/startup-count"] == "2"
    assert deployment.metadata.annotations["mindroom.ai/last-started-at"] == "11.0"
    assert [event.phase for event in events] == ["failed"]


def test_kubernetes_backend_ready_metadata_patch_failure_is_normalized(tmp_path: Path) -> None:
    """A failed ready-status patch should fail ensure_worker without tearing down a healthy worker."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    initial_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot = deepcopy(_TEST_TOOL_VALIDATION_SNAPSHOT)
    updated_snapshot["search"] = {
        "config_fields": ["engine"],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    }
    backend, apps_api, core_api = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=initial_snapshot,
    )
    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    updated_backend, _, _ = _backend(
        runtime_paths=runtime_paths,
        tool_validation_snapshot=updated_snapshot,
    )
    updated_backend._resources.apps_api = apps_api
    updated_backend._resources.core_api = core_api
    updated_backend._resources.api_exception_cls = _FakeApiError
    _install_real_elapsed_wait_for_ready(updated_backend, ready_after_seconds=1.6)

    original_patch_deployment = updated_backend._resources.patch_deployment

    def patch_deployment_with_ready_failure(
        deployment_name: str,
        *,
        replicas: int | None = None,
        annotations: dict[str, str] | None = None,
    ) -> None:
        if annotations is not None and annotations.get("mindroom.ai/worker-status") == "ready":
            msg = "ready patch failed"
            raise RuntimeError(msg)
        original_patch_deployment(
            deployment_name,
            replicas=replicas,
            annotations=annotations,
        )

    updated_backend._resources.patch_deployment = patch_deployment_with_ready_failure

    events: list[WorkerReadyProgress] = []
    with pytest.raises(WorkerBackendError, match="ready patch failed"):
        updated_backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=11.0,
            progress_sink=events.append,
        )

    worker_id = next(iter(apps_api.deployments))
    deployment = apps_api.deployments[worker_id]
    assert deployment.spec.replicas == 1
    assert worker_id in core_api.services
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] != "failed"
    assert [event.phase for event in events] == ["cold_start", "failed"]


@pytest.mark.parametrize(
    ("failure_target", "error_message"),
    [
        ("apply_deployment", "deployment reconcile failed"),
        ("apply_service", "service apply failed"),
    ],
)
def test_kubernetes_backend_warm_reconcile_failures_are_non_destructive(
    tmp_path: Path,
    failure_target: str,
    error_message: str,
) -> None:
    """Warm reconcile failures should fail ensure_worker without tearing down the existing worker."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths, idle_timeout_seconds=600.0)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)

    if failure_target == "apply_deployment":

        def apply_deployment_with_failure(**_kwargs: object) -> object:
            msg = error_message
            raise RuntimeError(msg)

        backend._resources.apply_deployment = apply_deployment_with_failure
    else:

        def apply_service_with_failure(worker_id: str) -> None:
            _ = worker_id
            msg = error_message
            raise RuntimeError(msg)

        backend._resources.apply_service = apply_service_with_failure

    events: list[WorkerReadyProgress] = []
    with pytest.raises(WorkerBackendError, match=error_message):
        backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=310.0,
            progress_sink=events.append,
        )

    deployment = apps_api.deployments[handle.worker_id]
    assert deployment.spec.replicas == 1
    assert handle.worker_id in core_api.services
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] != "failed"
    assert [event.phase for event in events] == []


@pytest.mark.parametrize(
    ("failure_target", "error_message"),
    [
        ("apply_service", "service apply failed"),
        ("wait_for_ready", "worker never became ready"),
    ],
)
def test_kubernetes_backend_existing_starting_worker_failures_record_failure(
    tmp_path: Path,
    failure_target: str,
    error_message: str,
) -> None:
    """Existing starting workers should still be marked failed when recovery fails."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, core_api = _backend(runtime_paths=runtime_paths, idle_timeout_seconds=600.0)
    handle = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    deployment = apps_api.deployments[handle.worker_id]
    deployment.metadata.annotations["mindroom.ai/worker-status"] = "starting"
    deployment.metadata.annotations.pop("mindroom.ai/failure-reason", None)
    deployment.status.ready_replicas = 0

    if failure_target == "apply_service":

        def apply_service_with_failure(worker_id: str) -> None:
            _ = worker_id
            msg = error_message
            raise RuntimeError(msg)

        backend._resources.apply_service = apply_service_with_failure
    else:

        def wait_for_ready_with_failure(
            _worker_id: str,
            *,
            timeout_seconds: float,
            deployment_ready_fn: object,
            on_poll_tick: object | None = None,
        ) -> object:
            del timeout_seconds, deployment_ready_fn, on_poll_tick
            msg = error_message
            raise WorkerBackendError(msg)

        backend._resources.wait_for_ready = wait_for_ready_with_failure

    events: list[WorkerReadyProgress] = []
    with pytest.raises(WorkerBackendError, match=error_message):
        backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=310.0,
            progress_sink=events.append,
        )

    deployment = apps_api.deployments[handle.worker_id]
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] == "failed"
    assert deployment.metadata.annotations["mindroom.ai/failure-reason"] == error_message
    assert deployment.spec.replicas == 0
    assert handle.worker_id not in core_api.services
    assert handle.worker_id not in core_api.secrets
    assert events[-1].phase == "failed"


@pytest.mark.parametrize(
    ("failure_target", "error_message"),
    [
        ("sync_shared_credentials", "sync credentials failed"),
        ("apply_service", "service apply failed"),
    ],
)
def test_kubernetes_backend_prestartup_failures_are_normalized(
    tmp_path: Path,
    failure_target: str,
    error_message: str,
) -> None:
    """Fresh startup failures before readiness polling should still persist failed worker state."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

    if failure_target == "sync_shared_credentials":
        failure_context = patch(
            "mindroom.workers.backends.kubernetes.sync_shared_credentials_to_worker",
            side_effect=RuntimeError(error_message),
        )
    else:

        def apply_service_with_failure(worker_id: str) -> None:
            _ = worker_id
            msg = error_message
            raise RuntimeError(msg)

        backend._resources.apply_service = apply_service_with_failure
        failure_context = nullcontext()

    events: list[WorkerReadyProgress] = []
    with failure_context, pytest.raises(WorkerBackendError, match=error_message):
        backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=10.0,
            progress_sink=events.append,
        )

    worker_id = backend._worker_id(_TEST_SCOPED_WORKER_KEY_A)
    deployment = apps_api.deployments[worker_id]
    assert deployment.metadata.annotations["mindroom.ai/worker-status"] == "failed"
    assert deployment.metadata.annotations["mindroom.ai/failure-reason"] == error_message
    assert deployment.metadata.annotations["mindroom.ai/startup-count"] == "1"
    assert deployment.metadata.annotations["mindroom.ai/last-started-at"] == "10.0"
    assert [event.phase for event in events] == ["failed"]


def test_kubernetes_backend_apply_deployment_failure_is_normalized(tmp_path: Path) -> None:
    """Deployment apply failures should surface as WorkerBackendError and emit failed progress."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, _apps_api, core_api = _backend(runtime_paths=runtime_paths)

    def apply_deployment_with_failure(**_kwargs: object) -> object:
        msg = "deployment apply failed"
        raise RuntimeError(msg)

    backend._resources.apply_deployment = apply_deployment_with_failure

    events: list[WorkerReadyProgress] = []
    with pytest.raises(WorkerBackendError, match="deployment apply failed"):
        backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=10.0,
            progress_sink=events.append,
        )

    assert [event.phase for event in events] == ["failed"]
    assert backend._worker_id(_TEST_SCOPED_WORKER_KEY_A) not in core_api.secrets


def test_kubernetes_backend_reports_failed_cold_start_progress() -> None:
    """Failed cold starts should surface a terminal failed progress event with the error."""
    backend, _apps_api, _core_api = _backend()
    events: list[WorkerReadyProgress] = []
    error_message = "worker never became ready"

    def _boom(
        _self: object,
        _deployment_name: str,
        *,
        timeout_seconds: float,
        deployment_ready_fn: object,
        on_poll_tick: Callable[[float], None] | None = None,
    ) -> object:
        del deployment_ready_fn
        assert on_poll_tick is not None
        started_at = time.monotonic()
        deadline = started_at + timeout_seconds
        while True:
            elapsed_seconds = time.monotonic() - started_at
            if elapsed_seconds >= 2.0:
                raise WorkerBackendError(error_message)
            assert time.monotonic() < deadline
            on_poll_tick(elapsed_seconds)
            time.sleep(0.01)

    backend._resources.wait_for_ready = MethodType(_boom, backend._resources)

    with pytest.raises(WorkerBackendError, match=error_message):
        backend.ensure_worker(
            WorkerSpec(_TEST_SCOPED_WORKER_KEY_A),
            now=10.0,
            progress_sink=events.append,
        )

    assert [event.phase for event in events] == ["cold_start", "failed"]
    assert events[-1].error == error_message


def test_kubernetes_backend_ignores_progress_when_sink_is_absent() -> None:
    """Cold starts should still succeed when the first caller has no sink."""
    backend, _apps_api, _core_api = _backend()

    def _ready(
        self: object,
        deployment_name: str,
        *,
        timeout_seconds: float,
        deployment_ready_fn: object,
        on_poll_tick: object | None = None,
    ) -> object:
        del timeout_seconds, deployment_ready_fn
        assert on_poll_tick is not None
        return self.read_deployment(deployment_name)

    backend._resources.wait_for_ready = MethodType(_ready, backend._resources)

    backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0, progress_sink=None)


def test_kubernetes_backend_replays_progress_snapshot_to_late_joining_waiter() -> None:
    """A waiter that joins after cold_start should replay that state before terminal ready."""
    backend, _apps_api, _core_api = _backend()
    worker_key = _TEST_SCOPED_WORKER_KEY_A
    first_events: list[WorkerReadyProgress] = []
    second_events: list[WorkerReadyProgress] = []
    cold_start_seen = threading.Event()
    ready_gate = threading.Event()
    second_registered = threading.Event()
    errors: list[BaseException] = []
    handles: dict[str, object] = {}

    def first_sink(progress: WorkerReadyProgress) -> None:
        first_events.append(progress)
        if progress.phase == "cold_start":
            cold_start_seen.set()

    def second_sink(progress: WorkerReadyProgress) -> None:
        second_events.append(progress)

    original_register = backend._register_progress_sink

    def register_with_signal(current_worker_key: str, progress_sink: object) -> None:
        original_register(current_worker_key, progress_sink)
        if current_worker_key == worker_key and progress_sink is second_sink:
            second_registered.set()

    backend._register_progress_sink = register_with_signal

    _install_real_elapsed_wait_for_ready(
        backend,
        ready_after_seconds=1.6,
        ready_gate=ready_gate,
    )

    def ensure_worker(name: str, *, progress_sink: Callable[[WorkerReadyProgress], None], now: float) -> None:
        try:
            handles[name] = backend.ensure_worker(
                WorkerSpec(worker_key),
                now=now,
                progress_sink=progress_sink,
            )
        except BaseException as exc:  # pragma: no cover - raised explicitly below
            errors.append(exc)

    first_thread = threading.Thread(
        target=ensure_worker,
        args=("first",),
        kwargs={"progress_sink": first_sink, "now": 10.0},
    )
    second_thread = threading.Thread(
        target=ensure_worker,
        args=("second",),
        kwargs={"progress_sink": second_sink, "now": 11.0},
    )

    first_thread.start()
    assert cold_start_seen.wait(timeout=3.0)
    second_thread.start()
    assert second_registered.wait(timeout=1.0)
    ready_gate.set()
    first_thread.join()
    second_thread.join()

    if errors:
        raise errors[0]

    assert [event.phase for event in first_events] == ["cold_start", "ready"]
    assert [event.phase for event in second_events] == ["cold_start", "ready"]
    assert handles["first"].worker_id == handles["second"].worker_id


def test_kubernetes_backend_keeps_digest_when_worker_name_prefix_is_long() -> None:
    """Long prefixes must still preserve the per-worker digest so names remain unique."""
    long_prefix = "mindroom-worker-prefix-that-is-intentionally-way-too-long-for-a-kubernetes-name"
    backend, _apps_api, _core_api = _backend(name_prefix=long_prefix)

    first = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_A), now=10.0)
    second = backend.ensure_worker(WorkerSpec(_TEST_SCOPED_WORKER_KEY_B), now=20.0)

    assert first.worker_id != second.worker_id
    assert len(first.worker_id) <= 63
    assert len(second.worker_id) <= 63


def _test_agent_vault_config(**overrides: object) -> KubernetesAgentVaultConfig:
    values: dict[str, object] = {
        "vault_name_prefix": "agent-vault",
        "cli_image": "example.test/agent-vault:1",
        "api_url": "http://agent-vault:14321",
        "proxy_url": "http://agent-vault:14322",
        "owner_email": "vault-owner@example.test",
        "bootstrap_secret_name": "agent-vault-bootstrap",
    }
    values.update(overrides)
    return KubernetesAgentVaultConfig(**values)  # type: ignore[arg-type]


def test_kubernetes_backend_adds_agent_vault_mint_init_container(tmp_path: Path) -> None:
    """An enabled worker pod gets a mint init container, token volume, and proxy env."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        agent_vault=_test_agent_vault_config(worker_ca_configmap_name="agent-vault-ca"),
    )
    worker_key = _TEST_SCOPED_WORKER_KEY_A

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

    deployment = next(b for b in apps_api.created_bodies if b["metadata"]["name"] == handle.worker_id)
    template_spec = deployment["spec"]["template"]["spec"]

    init = template_spec["initContainers"]
    assert len(init) == 1
    mint = init[0]
    assert mint["name"] == "agent-vault-mint-token"
    assert mint["image"] == "example.test/agent-vault:1"
    mint_script = mint["command"][2]
    assert "agent-vault owner vault join" in mint_script
    assert "agent-vault agent create" in mint_script
    assert "agent-vault agent rotate" in mint_script
    assert "agent-vault vault agent add" in mint_script
    assert "agent-vault vault agent set-role" in mint_script
    assert mint_script.index("agent-vault owner vault join") < mint_script.index("agent-vault agent create")
    assert "--role proxy > /dev/null 2>&1" not in mint_script
    assert ":proxy" in mint_script
    # The owner CLI session must not land on the shared token volume, or the
    # agent container (which mounts it) could read the owner credential.
    assert "export HOME=/tmp/agent-vault-mint-home" in mint_script
    assert 'export HOME="/agent-vault"' not in mint_script
    mint_env = {e["name"]: e["value"] for e in mint["env"]}
    # The vault name is the worker's own deterministic vault, owner email is configured.
    assert mint_env["AGENT_VAULT_VAULT"] == descriptive_worker_id_for_key(worker_key, prefix="agent-vault")
    assert mint_env["AGENT_VAULT_OWNER_EMAIL"] == "vault-owner@example.test"
    # The owner password (bootstrap secret) is mounted only on the init container.
    assert any(m["name"] == "agent-vault-bootstrap" for m in mint["volumeMounts"])

    main = template_spec["containers"][0]
    assert all(m["name"] != "agent-vault-bootstrap" for m in main["volumeMounts"])
    assert any(m["name"] == "agent-vault-token" and m.get("readOnly") for m in main["volumeMounts"])
    main_env = {e["name"]: e.get("value") for e in main["env"]}
    expected_vault = descriptive_worker_id_for_key(worker_key, prefix="agent-vault")
    assert main_env["MINDROOM_WORKER_EGRESS_PROXY_URL"] == "http://agent-vault:14322"
    assert main_env["MINDROOM_WORKER_EGRESS_PROXY_TOKEN_FILE"] == "/agent-vault/token"  # noqa: S105
    assert main_env["MINDROOM_WORKER_EGRESS_PROXY_VAULT"] == expected_vault
    assert main_env["MINDROOM_WORKER_EGRESS_PROXY_CA_FILE"] == "/etc/agent-vault/ca.pem"

    volume_names = {v["name"] for v in template_spec["volumes"]}
    assert {"agent-vault-token", "agent-vault-bootstrap", "agent-vault-ca"} <= volume_names

    # No bridge/NetworkPolicy resources exist in this model.
    assert backend._resources._agent_vault_vault_name(worker_key) == expected_vault


def test_agent_vault_main_env_error_names_worker_key_when_vault_name_missing(tmp_path: Path) -> None:
    """The defensive vault-name guard should include the worker key that failed."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, _apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        agent_vault=_test_agent_vault_config(),
    )
    worker_key = _TEST_SCOPED_WORKER_KEY_A

    def _missing_vault_name(_self: object, requested_worker_key: str) -> None:
        assert requested_worker_key == worker_key

    backend._resources._agent_vault_vault_name = MethodType(_missing_vault_name, backend._resources)

    with pytest.raises(WorkerBackendError) as exc_info:
        backend._resources._agent_vault_main_env(worker_key=worker_key)

    assert f"worker_key={worker_key!r}" in str(exc_info.value)


def test_kubernetes_backend_omits_agent_vault_when_disabled(tmp_path: Path) -> None:
    """Without Agent Vault config, no init container, token volume, or proxy env."""
    runtime_paths = resolve_primary_runtime_paths(
        config_path=Path("config.yaml"),
        storage_path=tmp_path / "mindroom-test-storage",
    )
    backend, apps_api, _core_api = _backend(runtime_paths=runtime_paths)

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

    deployment = next(b for b in apps_api.created_bodies if b["metadata"]["name"] == handle.worker_id)
    template_spec = deployment["spec"]["template"]["spec"]
    assert "initContainers" not in template_spec
    assert all(v["name"] != "agent-vault-token" for v in template_spec["volumes"])
    main_env = {e["name"] for e in template_spec["containers"][0]["env"]}
    assert "MINDROOM_WORKER_EGRESS_PROXY_URL" not in main_env
    assert backend._resources._agent_vault_vault_name(_TEST_SCOPED_WORKER_KEY_A) is None


def test_agent_vault_config_from_env_defaults_and_requirements() -> None:
    """Agent Vault env parsing applies defaults and enforces required fields."""
    assert KubernetesAgentVaultConfig.from_env({}) is None
    with pytest.raises(WorkerBackendError, match="CLI_IMAGE"):
        KubernetesAgentVaultConfig.from_env({"MINDROOM_KUBERNETES_AGENT_VAULT_ENABLED": "true"})
    with pytest.raises(WorkerBackendError, match="OWNER_EMAIL"):
        KubernetesAgentVaultConfig.from_env(
            {
                "MINDROOM_KUBERNETES_AGENT_VAULT_ENABLED": "true",
                "MINDROOM_KUBERNETES_AGENT_VAULT_CLI_IMAGE": "example.test/agent-vault:1",
            },
        )
    config = KubernetesAgentVaultConfig.from_env(
        {
            "MINDROOM_KUBERNETES_AGENT_VAULT_ENABLED": "true",
            "MINDROOM_KUBERNETES_AGENT_VAULT_CLI_IMAGE": "example.test/agent-vault:1",
            "MINDROOM_KUBERNETES_AGENT_VAULT_OWNER_EMAIL": "vault-owner@example.test",
        },
    )
    assert config is not None
    assert config.vault_name_prefix == "agent-vault"
    assert config.api_url == "http://agent-vault:14321"
    assert config.proxy_url == "http://agent-vault:14322"
    assert config.bootstrap_secret_name == "agent-vault-bootstrap"  # noqa: S105


def test_kubernetes_backend_config_from_runtime_reads_agent_vault(tmp_path: Path) -> None:
    """Backend config picks Agent Vault settings up from runtime env files."""
    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=kubernetes\n"
            "MINDROOM_KUBERNETES_WORKER_IMAGE=test-image\n"
            "MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME=test-pvc\n"
            "MINDROOM_KUBERNETES_AGENT_VAULT_ENABLED=true\n"
            "MINDROOM_KUBERNETES_AGENT_VAULT_CLI_IMAGE=example.test/agent-vault:1\n"
            "MINDROOM_KUBERNETES_AGENT_VAULT_OWNER_EMAIL=vault-owner@example.test\n"
        ),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(config_path=config_path)

    config = KubernetesWorkerBackendConfig.from_runtime(runtime_paths)

    assert config.agent_vault is not None
    assert config.agent_vault.cli_image == "example.test/agent-vault:1"
    assert config.agent_vault.owner_email == "vault-owner@example.test"
    signature = kubernetes_backend_config_signature(runtime_paths, auth_token="token")  # noqa: S106
    assert config.agent_vault.signature() in signature


def test_scoped_storage_policy_planning_resolves_config_includes(tmp_path: Path) -> None:
    """Committed Kubernetes config snapshots preserve resolved include data."""
    config_dir = tmp_path / "conf"
    config_dir.mkdir()
    (config_dir / "agents.yaml").write_text(
        "code:\n  display_name: Code\n  role: Writes code\n  worker_scope: user\n",
        encoding="utf-8",
    )
    config_path = config_dir / "config.yaml"
    config_path.write_text(
        """
agents: !include agents.yaml
models:
  default:
    provider: openai
    id: gpt-5.6
router:
  model: default
""".lstrip(),
        encoding="utf-8",
    )
    runtime_paths = resolve_primary_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )

    config_snapshot = serialized_kubernetes_worker_config_snapshot(load_config(runtime_paths))
    backend, _apps_api, _core_api = _backend(
        runtime_paths=runtime_paths,
        config_snapshot=config_snapshot,
    )

    assert backend._resources.resolved_agent_policies["code"].effective_execution_scope == "user"
