"""Shared constants and runtime path helpers for the mindroom package."""

import hashlib
import json
import os
import re
import shutil
import sys
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from types import MappingProxyType
from typing import TypeGuard, cast
from urllib.parse import quote

from dotenv import dotenv_values

from mindroom import runtime_env_policy

# Agent names
ROUTER_AGENT_NAME = "router"
VISIBLE_ROUTER_VOICE_ECHO_KEY = "com.mindroom.visible_router_voice_echo"
CONFIG_CONFIRMATION_REACTION_KEY = "com.mindroom.config_confirmation_reaction"
DEFAULT_MINDROOM_URL = "http://127.0.0.1:8765"
DEFAULT_COMPACTION_TIMEOUT_SECONDS = 600.0
DEFAULT_TOOL_OUTPUT_AUTO_SAVE_THRESHOLD_BYTES = 50 * 1024
KNOWLEDGE_FILE_INDEX_CONCURRENCY_ENV = "MINDROOM_KNOWLEDGE_FILE_INDEX_CONCURRENCY"
DEFAULT_MAX_CONCURRENT_KNOWLEDGE_FILE_INDEXES = 4
MAX_ALLOWED_CONCURRENT_KNOWLEDGE_FILE_INDEXES = 128
_MINDROOM_DISPATCH_THREAD_READ_TIMEOUT_SECONDS = 1.0
_STANDARD_HISTORY_ROLES = frozenset({"user", "assistant", "tool"})
_PROMPT_HISTORY_STORAGE_ROLES = frozenset({"system", "developer"})

# Search order for existing files: env var > ./config.yaml > ~/.mindroom/config.yaml
_CONFIG_SEARCH_PATHS = [Path("config.yaml"), Path.home() / ".mindroom" / "config.yaml"]
CONTROL_STATE_PATH_ENV = runtime_env_policy.CONTROL_STATE_PATH_ENV
_RUNTIME_PATH_ENV_KEYS = frozenset({"MINDROOM_CONFIG_PATH", "MINDROOM_STORAGE_PATH", CONTROL_STATE_PATH_ENV})
_SANDBOX_STARTUP_MANIFEST_RELATIVE_PATH = Path(".runtime") / "startup_manifest.json"
_CONFIG_PATH_PLACEHOLDER_PATTERN = re.compile(r"\$(?:\{(?P<braced>[A-Z0-9_]+)\}|(?P<bare>[A-Z0-9_]+))")

# Bash bookkeeping vars that change every time printenv runs and are never
# meaningful overlay output from `.mindroom/worker-env.sh`.
WORKSPACE_ENV_OVERLAY_TRANSIENT_NAMES = frozenset({"PWD", "OLDPWD", "SHLVL", "_", "PIPESTATUS"})
_WORKSPACE_HOME_IDENTITY_ENV_NAMES = frozenset(
    {
        "HOME",
        "MINDROOM_AGENT_WORKSPACE",
        "XDG_CONFIG_HOME",
        "XDG_DATA_HOME",
        "XDG_STATE_HOME",
    },
)
WORKER_RUNTIME_PATH_ENV_NAMES = frozenset(
    {
        "XDG_CACHE_HOME",
        "PIP_CACHE_DIR",
        "UV_CACHE_DIR",
        "PYTHONPYCACHEPREFIX",
        "VIRTUAL_ENV",
    },
)
WORKSPACE_HOME_CONTRACT_ENV_NAMES = _WORKSPACE_HOME_IDENTITY_ENV_NAMES | WORKER_RUNTIME_PATH_ENV_NAMES


def prompt_roles_for_history_storage(system_message_role: str = "system") -> frozenset[str]:
    """Return prompt roles that should be stripped before durable history storage."""
    roles = set(_PROMPT_HISTORY_STORAGE_ROLES)
    configured_role = system_message_role.strip()
    if configured_role and configured_role not in _STANDARD_HISTORY_ROLES:
        roles.add(configured_role)
    return frozenset(roles)


def workspace_home_identity_env(workspace: Path | str) -> dict[str, str]:
    """Build the workspace HOME identity env fragment."""
    workspace_text = str(workspace)
    workspace_path = Path(workspace_text)
    return {
        "HOME": workspace_text,
        "MINDROOM_AGENT_WORKSPACE": workspace_text,
        "XDG_CONFIG_HOME": str(workspace_path / ".config"),
        "XDG_DATA_HOME": str(workspace_path / ".local" / "share"),
        "XDG_STATE_HOME": str(workspace_path / ".local" / "state"),
    }


def subprocess_path_with_prepends(
    current_path: str | None,
    *,
    prepend_entries: tuple[str, ...] = (),
) -> str | None:
    """Return a PATH value with prepended entries first and duplicate entries removed."""
    if current_path is None and not prepend_entries:
        return current_path

    path_entries = [entry for entry in prepend_entries if entry]
    if current_path:
        path_entries.extend(entry for entry in current_path.split(os.pathsep) if entry)

    if not path_entries:
        return current_path

    deduped_entries: list[str] = []
    seen_entries: set[str] = set()
    for entry in path_entries:
        if entry in seen_entries:
            continue
        seen_entries.add(entry)
        deduped_entries.append(entry)
    return os.pathsep.join(deduped_entries)


def is_workspace_env_overlay_name_allowed(name: str) -> bool:
    """Return whether one env var name may be returned from `.mindroom/worker-env.sh`.

    The agent-editable overlay only protects runner control-plane names.
    Other explicitly exported values are user intent and pass through.
    """
    if not name:
        return False
    return runtime_env_policy.is_shell_passthrough_allowed_env_name(name)


@dataclass(frozen=True)
class RuntimePaths:
    """Resolved runtime context shared across the process.

    `RuntimePaths` is the source of truth for:
    - active config path and config dir
    - config-adjacent `.env`
    - shared storage root
    - the true exported process env seen during resolution
    - the sibling `.env` values seen during resolution

    Runtime env precedence is:
    1. Explicit runtime arguments passed to `resolve_runtime_paths()`
    2. Exported process env values
    3. The config-adjacent `.env`
    4. Code defaults in the caller
    """

    config_path: Path
    config_dir: Path
    env_path: Path
    storage_root: Path
    control_state_root: Path | None = None
    process_env: Mapping[str, str] = field(default_factory=dict, repr=False)
    env_file_values: Mapping[str, str] = field(default_factory=dict, repr=False)

    def env_value(self, name: str, *, default: str | None = None) -> str | None:
        """Resolve one env value against this runtime context."""
        if name == "MINDROOM_CONFIG_PATH":
            return str(self.config_path)
        if name == "MINDROOM_STORAGE_PATH":
            return str(self.storage_root)
        if name == CONTROL_STATE_PATH_ENV:
            return str(self.control_state_root) if self.control_state_root is not None else default
        if name in self.process_env:
            return self.process_env[name]
        if name in self.env_file_values:
            return self.env_file_values[name]
        return default

    def env_flag(self, name: str, *, default: bool = False) -> bool:
        """Resolve one boolean env value against this runtime context."""
        value = self.env_value(name)
        if value is None:
            return default
        return value.strip().lower() in {"1", "true", "yes", "on"}


def _copy_process_env(process_env: dict[str, str] | None = None) -> dict[str, str]:
    if process_env is not None:
        return dict(process_env)
    return dict(os.environ)


def _runtime_env_file_values_for_path(env_path: Path) -> dict[str, str]:
    """Read string env values from one runtime-adjacent `.env` file."""
    if not env_path.is_file():
        return {}
    return {key: value for key, value in dotenv_values(env_path).items() if isinstance(value, str)}


def _resolve_runtime_relative_path(raw_value: str, *, base_dir: Path) -> Path:
    """Resolve one runtime-owned path value relative to its config directory."""
    path = Path(raw_value).expanduser()
    if not path.is_absolute():
        path = base_dir / path
    return path.resolve()


def _configured_config_path(process_env: Mapping[str, str]) -> Path | None:
    configured_path = process_env.get("MINDROOM_CONFIG_PATH", "").strip()
    if not configured_path:
        return None
    return Path(configured_path).expanduser()


def config_search_locations(process_env: Mapping[str, str]) -> list[Path]:
    """Return the ordered list of locations where MindRoom looks for config.

    This is the single source of truth for config file discovery.
    """
    seen: set[Path] = set()
    locations: list[Path] = []
    if configured_path := _configured_config_path(process_env):
        resolved = configured_path.resolve()
        seen.add(resolved)
        locations.append(resolved)
    for p in _CONFIG_SEARCH_PATHS:
        resolved = p.resolve()
        if resolved not in seen:
            seen.add(resolved)
            locations.append(resolved)
    return locations


def _storage_root_from_env_values(env_file_values: dict[str, str], *, config_dir: Path) -> Path | None:
    value = env_file_values.get("MINDROOM_STORAGE_PATH")
    if value is None or not value.strip():
        return None
    return _resolve_runtime_relative_path(value, base_dir=config_dir)


def _control_state_root_from_env_values(env_values: Mapping[str, str], *, config_dir: Path) -> Path | None:
    value = env_values.get(CONTROL_STATE_PATH_ENV)
    if value is None or not value.strip():
        return None
    return _resolve_runtime_relative_path(value, base_dir=config_dir)


def _default_control_state_root(storage_root: Path) -> Path:
    return (storage_root / "control_state").resolve()


def resolve_runtime_paths(
    *,
    config_path: Path | None = None,
    storage_path: Path | None = None,
    process_env: dict[str, str] | None = None,
) -> RuntimePaths:
    """Resolve the runtime config/env/storage paths for one execution context.

    This is a pure resolver. It does not mutate `os.environ` or any module globals.
    """
    resolved_config_arg = Path(config_path).expanduser().resolve() if config_path is not None else None
    resolved_process_env = _copy_process_env(process_env)
    resolved_config_path = (
        Path(resolved_config_arg or _find_config(process_env=resolved_process_env)).expanduser().resolve()
    )
    config_dir = resolved_config_path.parent
    env_path = config_dir / ".env"
    env_file_values = _runtime_env_file_values_for_path(env_path)

    configured_storage_root = resolved_process_env.get("MINDROOM_STORAGE_PATH", "").strip()
    configured_storage_path = Path(configured_storage_root).expanduser().resolve() if configured_storage_root else None

    if storage_path is not None:
        resolved_storage_root = Path(storage_path).expanduser().resolve()
    elif configured_storage_path is not None:
        resolved_storage_root = configured_storage_path
    elif env_storage_root := _storage_root_from_env_values(env_file_values, config_dir=config_dir):
        resolved_storage_root = env_storage_root
    else:
        resolved_storage_root = (config_dir / "mindroom_data").resolve()
    resolved_control_state_root = _control_state_root_from_env_values(
        resolved_process_env,
        config_dir=config_dir,
    )
    if resolved_control_state_root is None:
        resolved_control_state_root = _control_state_root_from_env_values(
            env_file_values,
            config_dir=config_dir,
        )
    if resolved_control_state_root is None:
        resolved_control_state_root = _default_control_state_root(resolved_storage_root)

    return RuntimePaths(
        config_path=resolved_config_path,
        config_dir=config_dir,
        env_path=env_path,
        storage_root=resolved_storage_root,
        control_state_root=resolved_control_state_root,
        process_env=cast("Mapping[str, str]", MappingProxyType(resolved_process_env)),
        env_file_values=cast("Mapping[str, str]", MappingProxyType(env_file_values)),
    )


def _with_primary_runtime_env(paths: RuntimePaths) -> RuntimePaths:
    """Return one runtime context whose process env snapshot carries its path contract."""
    normalized_process_env = dict(paths.process_env)
    normalized_process_env["MINDROOM_CONFIG_PATH"] = str(paths.config_path)
    normalized_process_env["MINDROOM_STORAGE_PATH"] = str(paths.storage_root)
    if normalized_process_env == dict(paths.process_env):
        return paths
    return RuntimePaths(
        config_path=paths.config_path,
        config_dir=paths.config_dir,
        env_path=paths.env_path,
        storage_root=paths.storage_root,
        control_state_root=paths.control_state_root,
        process_env=cast("Mapping[str, str]", MappingProxyType(normalized_process_env)),
        env_file_values=paths.env_file_values,
    )


def resolve_primary_runtime_paths(
    *,
    config_path: Path | None = None,
    storage_path: Path | None = None,
    process_env: dict[str, str] | None = None,
) -> RuntimePaths:
    """Resolve the primary runtime context for one top-level execution boundary."""
    return _with_primary_runtime_env(
        resolve_runtime_paths(
            config_path=config_path,
            storage_path=storage_path,
            process_env=process_env,
        ),
    )


def serialize_runtime_paths(runtime_paths: RuntimePaths) -> dict[str, object]:
    """Return a JSON-compatible payload for explicit cross-process runtime handoff."""
    return {
        "config_path": str(runtime_paths.config_path),
        "storage_root": str(runtime_paths.storage_root),
        "process_env": dict(runtime_paths.process_env),
        "env_file_values": dict(runtime_paths.env_file_values),
    }


def serialize_public_runtime_paths(runtime_paths: RuntimePaths) -> dict[str, object]:
    """Return a JSON payload for pod-visible worker startup without secrets."""
    process_env = runtime_env_policy.public_worker_startup_env(runtime_paths.process_env)
    env_file_values = runtime_env_policy.public_worker_startup_env(runtime_paths.env_file_values)
    return {
        "config_path": str(runtime_paths.config_path),
        "storage_root": str(runtime_paths.storage_root),
        "process_env": process_env,
        "env_file_values": env_file_values,
    }


def _serialize_startup_manifest(
    runtime_paths: RuntimePaths,
    *,
    tool_validation_snapshot: Mapping[str, object] | None = None,
    public_runtime: bool = False,
) -> dict[str, object]:
    """Return one JSON-compatible startup manifest for sandbox runners."""
    return {
        "runtime_paths": serialize_public_runtime_paths(runtime_paths)
        if public_runtime
        else serialize_runtime_paths(runtime_paths),
        "tool_validation_snapshot": dict(tool_validation_snapshot or {}),
    }


def _startup_manifest_json(
    runtime_paths: RuntimePaths,
    *,
    tool_validation_snapshot: Mapping[str, object] | None = None,
    public_runtime: bool = False,
) -> str:
    """Return one deterministic JSON string for sandbox-runner startup state."""
    return json.dumps(
        _serialize_startup_manifest(
            runtime_paths,
            tool_validation_snapshot=tool_validation_snapshot,
            public_runtime=public_runtime,
        ),
        separators=(",", ":"),
        sort_keys=True,
    )


def startup_manifest_sha256(
    runtime_paths: RuntimePaths,
    *,
    tool_validation_snapshot: Mapping[str, object] | None = None,
    public_runtime: bool = False,
) -> str:
    """Return one stable content hash for sandbox-runner startup state."""
    payload = _startup_manifest_json(
        runtime_paths,
        tool_validation_snapshot=tool_validation_snapshot,
        public_runtime=public_runtime,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def sandbox_startup_manifest_path(storage_root: Path) -> Path:
    """Return the canonical startup manifest path under one runtime root."""
    return storage_root / _SANDBOX_STARTUP_MANIFEST_RELATIVE_PATH


def write_startup_manifest(
    storage_root: Path,
    runtime_paths: RuntimePaths,
    *,
    tool_validation_snapshot: Mapping[str, object] | None = None,
    public_runtime: bool = False,
) -> Path:
    """Write one sandbox-runner startup manifest and return its path."""
    manifest_path = sandbox_startup_manifest_path(storage_root)
    manifest_path.parent.mkdir(parents=True, exist_ok=True)
    manifest_path.write_text(
        _startup_manifest_json(
            runtime_paths,
            tool_validation_snapshot=tool_validation_snapshot,
            public_runtime=public_runtime,
        ),
        encoding="utf-8",
    )
    return manifest_path


def _is_json_object(value: object) -> TypeGuard[dict[str, object]]:
    return isinstance(value, dict)


def deserialize_runtime_paths(payload: object) -> RuntimePaths:
    """Build one RuntimePaths object from an explicit serialized payload."""
    if not _is_json_object(payload):
        msg = "Serialized runtime payload must be a JSON object"
        raise TypeError(msg)
    raw_config_path = payload.get("config_path")
    raw_storage_root = payload.get("storage_root")
    raw_process_env = payload.get("process_env")
    raw_env_file_values = payload.get("env_file_values")
    if not isinstance(raw_config_path, str) or not raw_config_path.strip():
        msg = "Serialized runtime payload is missing config_path"
        raise TypeError(msg)
    if not isinstance(raw_storage_root, str) or not raw_storage_root.strip():
        msg = "Serialized runtime payload is missing storage_root"
        raise TypeError(msg)
    if not isinstance(raw_process_env, Mapping):
        msg = "Serialized runtime payload is missing process_env"
        raise TypeError(msg)
    if not isinstance(raw_env_file_values, Mapping):
        msg = "Serialized runtime payload is missing env_file_values"
        raise TypeError(msg)

    process_env = {
        key: value for key, value in raw_process_env.items() if isinstance(key, str) and isinstance(value, str)
    }
    env_file_values = {
        key: value for key, value in raw_env_file_values.items() if isinstance(key, str) and isinstance(value, str)
    }
    config_path = Path(raw_config_path).expanduser().resolve()
    storage_root = Path(raw_storage_root).expanduser().resolve()
    return RuntimePaths(
        config_path=config_path,
        config_dir=config_path.parent,
        env_path=config_path.parent / ".env",
        storage_root=storage_root,
        process_env=cast("Mapping[str, str]", MappingProxyType(process_env)),
        env_file_values=cast("Mapping[str, str]", MappingProxyType(env_file_values)),
    )


def deserialize_startup_manifest(payload: object) -> tuple[RuntimePaths, object]:
    """Build one startup manifest from explicit serialized payload."""
    if not _is_json_object(payload):
        msg = "Serialized startup manifest must be a JSON object"
        raise TypeError(msg)
    raw_runtime_paths = payload.get("runtime_paths")
    if raw_runtime_paths is None:
        msg = "Serialized startup manifest is missing runtime_paths"
        raise TypeError(msg)
    return deserialize_runtime_paths(raw_runtime_paths), payload.get("tool_validation_snapshot", {})


def _expand_runtime_path_vars(value: str, paths: RuntimePaths) -> str:
    """Expand the allowed config-path placeholders for one runtime context."""

    def _replace(match: re.Match[str]) -> str:
        name = match.group("braced") or match.group("bare") or ""
        if name == "MINDROOM_CONFIG_PATH":
            return str(paths.config_path)
        if name == "MINDROOM_STORAGE_PATH":
            return str(paths.storage_root)
        msg = (
            "Config-relative paths only support ${MINDROOM_CONFIG_PATH} and "
            f"${{MINDROOM_STORAGE_PATH}} placeholders (got: {name})"
        )
        raise ValueError(msg)

    return _CONFIG_PATH_PLACEHOLDER_PATTERN.sub(_replace, value)


def exported_process_env() -> dict[str, str]:
    """Return the current exported env snapshot."""
    return _copy_process_env()


def runtime_env_values(runtime_paths: RuntimePaths) -> Mapping[str, str]:
    """Return the effective runtime env mapping for one explicit runtime context."""
    merged_env = dict(runtime_paths.env_file_values)
    merged_env.update(runtime_paths.process_env)
    merged_env["MINDROOM_CONFIG_PATH"] = str(runtime_paths.config_path)
    merged_env["MINDROOM_STORAGE_PATH"] = str(runtime_paths.storage_root)
    return cast("Mapping[str, str]", MappingProxyType(merged_env))


def runtime_paths_with_config_path(runtime_paths: RuntimePaths, config_path: Path) -> RuntimePaths:
    """Return a primary-runtime context rebased to one explicit config path."""
    resolved_config_path = Path(config_path).expanduser().resolve()
    if resolved_config_path == runtime_paths.config_path:
        return _with_primary_runtime_env(runtime_paths)
    return resolve_primary_runtime_paths(
        config_path=resolved_config_path,
        storage_path=runtime_paths.storage_root,
        process_env=dict(runtime_paths.process_env),
    )


def runtime_paths_with_storage_root(runtime_paths: RuntimePaths, storage_root: Path) -> RuntimePaths:
    """Return a runtime context rebased to one explicit storage root."""
    resolved_storage_root = Path(storage_root).expanduser().resolve()
    normalized_process_env = dict(runtime_paths.process_env)
    normalized_process_env["MINDROOM_CONFIG_PATH"] = str(runtime_paths.config_path)
    normalized_process_env["MINDROOM_STORAGE_PATH"] = str(resolved_storage_root)
    old_default_control_state_root = _default_control_state_root(runtime_paths.storage_root)
    current_control_state_root = runtime_paths.control_state_root
    if current_control_state_root is None or current_control_state_root == old_default_control_state_root:
        resolved_control_state_root = _default_control_state_root(resolved_storage_root)
    else:
        resolved_control_state_root = current_control_state_root
    if resolved_storage_root == runtime_paths.storage_root and normalized_process_env == dict(
        runtime_paths.process_env,
    ):
        return runtime_paths
    return RuntimePaths(
        config_path=runtime_paths.config_path,
        config_dir=runtime_paths.config_dir,
        env_path=runtime_paths.env_path,
        storage_root=resolved_storage_root,
        control_state_root=resolved_control_state_root,
        process_env=cast("Mapping[str, str]", MappingProxyType(normalized_process_env)),
        env_file_values=runtime_paths.env_file_values,
    )


def _trusted_tool_runtime_env_layers(
    runtime_paths: RuntimePaths,
) -> tuple[dict[str, str], dict[str, str]]:
    env_file_values = {
        key: value
        for key, value in runtime_paths.env_file_values.items()
        if runtime_env_policy.is_trusted_tool_runtime_env_file_name(key)
    }
    process_env = {
        key: value
        for key, value in runtime_paths.process_env.items()
        if runtime_env_policy.is_trusted_tool_runtime_process_env_name(key)
    }
    return process_env, env_file_values


def _isolated_runtime_env_layers(
    runtime_paths: RuntimePaths,
) -> tuple[dict[str, str], dict[str, str]]:
    env_file_values = {
        key: value
        for key, value in runtime_paths.env_file_values.items()
        if runtime_env_policy.is_isolated_worker_runtime_env_name(key)
    }
    process_env = {
        key: value
        for key, value in runtime_paths.process_env.items()
        if runtime_env_policy.is_isolated_worker_runtime_env_name(key)
    }
    return process_env, env_file_values


def _shell_extra_env_patterns(extra_env_passthrough: str | None) -> tuple[str, ...]:
    if extra_env_passthrough is None:
        return ()
    return tuple(part for part in re.split(r"[\s,]+", extra_env_passthrough.strip()) if part)


def shell_extra_env_values(
    *,
    extra_env_passthrough: str | None = None,
    process_env: Mapping[str, str] | None = None,
) -> Mapping[str, str]:
    """Return explicit extra env values that shell execution may inherit."""
    patterns = _shell_extra_env_patterns(extra_env_passthrough)
    if not patterns:
        return cast("Mapping[str, str]", MappingProxyType({}))

    source_env = os.environ if process_env is None else process_env
    selected_env = runtime_env_policy.shell_passthrough_env(source_env, patterns=patterns)
    return cast("Mapping[str, str]", MappingProxyType(selected_env))


def _sandbox_shell_system_env_values(
    *,
    process_env: Mapping[str, str] | None = None,
) -> Mapping[str, str]:
    """Return the non-secret system env shell commands may receive by default."""
    source_env = os.environ if process_env is None else process_env
    return runtime_env_policy.sandbox_shell_system_env(source_env)


def trusted_tool_runtime_env_values(
    runtime_paths: RuntimePaths,
) -> Mapping[str, str]:
    """Return the runtime env available while trusted code rebuilds tool instances.

    This intentionally differs from ``runtime_env_values()``:
    - config-adjacent ``.env`` values remain visible to trusted tool construction
    - exported process env is filtered to the committed runtime contract
    - internal control env such as sandbox auth tokens stay excluded
    """
    process_env, env_file_values = _trusted_tool_runtime_env_layers(runtime_paths)
    merged_env = dict(env_file_values)
    merged_env.update(process_env)
    for name in tuple(merged_env):
        if name != "GOOGLE_APPLICATION_CREDENTIALS" and not name.endswith("_FILE"):
            continue
        source_path = _runtime_env_source_path(runtime_paths, name)
        if source_path is None:
            continue
        merged_env[name] = str(source_path.resolve())
    merged_env["MINDROOM_CONFIG_PATH"] = str(runtime_paths.config_path)
    merged_env["MINDROOM_STORAGE_PATH"] = str(runtime_paths.storage_root)
    return cast("Mapping[str, str]", MappingProxyType(merged_env))


def _execution_tool_runtime_env_values(runtime_paths: RuntimePaths) -> Mapping[str, str]:
    """Return the stricter env visible to sandbox-proxied execution tools."""
    process_env = runtime_env_policy.execution_tool_runtime_env(runtime_paths.process_env)
    env_file_values = runtime_env_policy.execution_tool_runtime_env(runtime_paths.env_file_values)
    merged_env = dict(env_file_values)
    merged_env.update(process_env)
    merged_env["MINDROOM_CONFIG_PATH"] = str(runtime_paths.config_path)
    merged_env["MINDROOM_STORAGE_PATH"] = str(runtime_paths.storage_root)
    return cast("Mapping[str, str]", MappingProxyType(merged_env))


def isolated_runtime_paths(runtime_paths: RuntimePaths) -> RuntimePaths:
    """Return one runtime view filtered for isolated worker execution."""
    process_env, env_file_values = _isolated_runtime_env_layers(runtime_paths)
    return RuntimePaths(
        config_path=runtime_paths.config_path,
        config_dir=runtime_paths.config_dir,
        env_path=runtime_paths.env_path,
        storage_root=runtime_paths.storage_root,
        control_state_root=None,
        process_env=cast("Mapping[str, str]", MappingProxyType(process_env)),
        env_file_values=cast("Mapping[str, str]", MappingProxyType(env_file_values)),
    )


def shell_execution_runtime_env_values(
    runtime_paths: RuntimePaths,
    *,
    extra_env_passthrough: str | None = None,
    process_env: Mapping[str, str] | None = None,
) -> Mapping[str, str]:
    """Return the env visible to shell execution after explicit passthrough is applied."""
    merged_env = dict(
        shell_extra_env_values(
            extra_env_passthrough=extra_env_passthrough,
            process_env=process_env,
        ),
    )
    merged_env.update(trusted_tool_runtime_env_values(runtime_paths))
    return cast("Mapping[str, str]", MappingProxyType(merged_env))


def _runtime_env_source_path(runtime_paths: RuntimePaths, name: str) -> Path | None:
    """Return one runtime env var as a lexical filesystem path without resolving symlinks."""
    raw_value = runtime_paths.env_value(name)
    if raw_value is None or not raw_value.strip():
        return None
    path = Path(raw_value).expanduser()
    if not path.is_absolute():
        path = runtime_paths.config_dir / path
    return path


def _sandbox_shell_execution_runtime_env_values(
    *,
    extra_env_passthrough: str | None = None,
    process_env: Mapping[str, str] | None = None,
) -> Mapping[str, str]:
    """Return the stricter env visible to sandbox-proxied shell execution."""
    merged_env = dict(_sandbox_shell_system_env_values(process_env=process_env))
    merged_env.update(
        shell_extra_env_values(
            extra_env_passthrough=extra_env_passthrough,
            process_env=process_env,
        ),
    )
    return cast("Mapping[str, str]", MappingProxyType(merged_env))


EXECUTION_ENV_TOOL_NAMES = frozenset({"python", "shell"})

# Worker pod env (set by the Kubernetes backend) that lets the runner compose
# the Agent Vault egress proxy for python/shell. The token never reaches the
# primary; it is minted into the worker pod and read here at execution time.
_WORKER_EGRESS_PROXY_URL_ENV = runtime_env_policy.WORKER_EGRESS_PROXY_ENV_BY_KEY["proxy_url"]
_WORKER_EGRESS_PROXY_TOKEN_FILE_ENV = runtime_env_policy.WORKER_EGRESS_PROXY_ENV_BY_KEY["token_file"]
_WORKER_EGRESS_PROXY_VAULT_ENV = runtime_env_policy.WORKER_EGRESS_PROXY_ENV_BY_KEY["vault"]
_WORKER_EGRESS_PROXY_CA_FILE_ENV = runtime_env_policy.WORKER_EGRESS_PROXY_ENV_BY_KEY["ca_file"]
_WORKER_EGRESS_NO_PROXY = "localhost,127.0.0.1,::1,.svc,.cluster.local"


def _git_config_env(process_env: Mapping[str, str], entries: list[tuple[str, str]]) -> dict[str, str]:
    """Append entries to any well-formed environment-backed Git config.

    A malformed incoming block is discarded so Git never receives an
    inconsistent count/key/value set.
    """
    existing_entries: list[tuple[str, str]] = []
    try:
        existing_count = int(process_env.get("GIT_CONFIG_COUNT", "0"))
    except ValueError:
        existing_count = 0
    if existing_count > 0:
        for index in range(existing_count):
            key = process_env.get(f"GIT_CONFIG_KEY_{index}")
            value = process_env.get(f"GIT_CONFIG_VALUE_{index}")
            if key is None or value is None:
                existing_entries = []
                break
            existing_entries.append((key, value))

    merged_entries = [*existing_entries, *entries]
    env = {"GIT_CONFIG_COUNT": str(len(merged_entries))}
    for index, (key, value) in enumerate(merged_entries):
        env[f"GIT_CONFIG_KEY_{index}"] = key
        env[f"GIT_CONFIG_VALUE_{index}"] = value
    return env


def worker_proxy_execution_env(process_env: Mapping[str, str]) -> dict[str, str]:
    """Return the per-worker egress proxy env overlay for python/shell, or ``{}``.

    General mechanism, provider-agnostic: reads a per-worker proxy endpoint plus
    a token file written into the worker pod and composes proxy basic-auth
    userinfo. The token becomes the username; when configured, the vault name
    becomes the password. Returns empty when no per-worker egress proxy is
    configured for this worker or the token is not yet present.
    """
    proxy_url = (process_env.get(_WORKER_EGRESS_PROXY_URL_ENV) or "").strip()
    token_file = (process_env.get(_WORKER_EGRESS_PROXY_TOKEN_FILE_ENV) or "").strip()
    if not proxy_url or "://" not in proxy_url or not token_file:
        return {}
    try:
        token = Path(token_file).read_text(encoding="utf-8").strip()
    except OSError:
        return {}
    if not token:
        return {}
    vault = (process_env.get(_WORKER_EGRESS_PROXY_VAULT_ENV) or "").strip()
    scheme, _, rest = proxy_url.partition("://")
    # Percent-encode proxy basic-auth userinfo so URL-significant characters
    # (@ : / # ? % +, whitespace) cannot make HTTP clients mis-parse auth.
    authed = f"{scheme}://{quote(token, safe='')}:{quote(vault, safe='')}@{rest}"
    env = {
        "HTTP_PROXY": authed,
        "HTTPS_PROXY": authed,
        "http_proxy": authed,
        "https_proxy": authed,
        # Git/libcurl does not always preemptively send proxy credentials from
        # HTTPS_PROXY userinfo on CONNECT. Environment-backed Git config makes
        # the same proxy explicit without writing user/global git config.
        "NO_PROXY": _WORKER_EGRESS_NO_PROXY,
        "no_proxy": _WORKER_EGRESS_NO_PROXY,
    }
    env.update(
        _git_config_env(
            process_env,
            [
                ("http.proxy", authed),
                ("http.proxyAuthMethod", "basic"),
            ],
        ),
    )
    ca_file = (process_env.get(_WORKER_EGRESS_PROXY_CA_FILE_ENV) or "").strip()
    if ca_file:
        env["REQUESTS_CA_BUNDLE"] = ca_file
        env["CURL_CA_BUNDLE"] = ca_file
        env["SSL_CERT_FILE"] = ca_file
        # git and node do not honor the bundles above: git needs GIT_SSL_CAINFO
        # (else `git clone` over HTTPS rejects the MITM proxy's certificate)
        # and node only *adds* roots via NODE_EXTRA_CA_CERTS.
        env["GIT_SSL_CAINFO"] = ca_file
        env["NODE_EXTRA_CA_CERTS"] = ca_file
    return env


def build_execution_tool_env(
    tool_name: str,
    runtime_paths: RuntimePaths,
    *,
    extra_env_passthrough: str | None = None,
    shell_process_env: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """Build the from-scratch execution env for an execution tool (shell or python).

    Callers gate on :data:`EXECUTION_ENV_TOOL_NAMES` first. The shell
    ``process_env`` source differs between the client proxy (which only trusts
    the runtime paths) and the worker runner (which may merge ``os.environ``),
    so it is supplied by the caller.
    """
    if tool_name == "shell":
        env = dict(
            _sandbox_shell_execution_runtime_env_values(
                extra_env_passthrough=extra_env_passthrough,
                process_env=shell_process_env,
            ),
        )
        env.update(worker_proxy_execution_env(shell_process_env or runtime_paths.process_env))
        return env
    env = dict(_execution_tool_runtime_env_values(runtime_paths))
    env.update(worker_proxy_execution_env(runtime_paths.process_env))
    return env


def runtime_env_path(runtime_paths: RuntimePaths, name: str) -> Path | None:
    """Resolve one runtime env var as a filesystem path.

    Relative paths are interpreted relative to the runtime config directory.
    """
    source_path = _runtime_env_source_path(runtime_paths, name)
    if source_path is None:
        return None
    return source_path.resolve()


def runtime_env_flag(
    name: str,
    runtime_paths: RuntimePaths,
    *,
    default: bool = False,
) -> bool:
    """Read a boolean runtime env flag with config-adjacent `.env` fallback."""
    value = runtime_paths.env_value(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def runtime_dispatch_thread_read_timeout_seconds(runtime_paths: RuntimePaths) -> float:
    """Return the dispatch-safe thread read wall-clock budget."""
    raw_value = runtime_paths.env_value("MINDROOM_DISPATCH_THREAD_READ_TIMEOUT_SECONDS")
    if raw_value is None:
        return _MINDROOM_DISPATCH_THREAD_READ_TIMEOUT_SECONDS
    try:
        return max(0.001, float(raw_value))
    except ValueError:
        return _MINDROOM_DISPATCH_THREAD_READ_TIMEOUT_SECONDS


def runtime_matrix_homeserver(runtime_paths: RuntimePaths) -> str:
    """Return the effective Matrix homeserver for one runtime context."""
    return runtime_paths.env_value("MATRIX_HOMESERVER", default="http://localhost:8008") or "http://localhost:8008"


def runtime_matrix_ssl_verify(runtime_paths: RuntimePaths) -> bool:
    """Return whether Matrix HTTPS requests should verify certificates."""
    return runtime_env_flag("MATRIX_SSL_VERIFY", default=True, runtime_paths=runtime_paths)


def runtime_matrix_server_name(runtime_paths: RuntimePaths) -> str | None:
    """Return the optional Matrix server-name override for one runtime context."""
    return runtime_paths.env_value("MATRIX_SERVER_NAME")


def runtime_mindroom_namespace(runtime_paths: RuntimePaths) -> str | None:
    """Return the optional installation namespace for one runtime context."""
    value = runtime_paths.env_value("MINDROOM_NAMESPACE")
    if value is None:
        return None
    normalized = value.strip().lower()
    return normalized or None


def matrix_state_file(runtime_paths: RuntimePaths) -> Path:
    """Return the matrix-state file for one runtime context."""
    return runtime_paths.storage_root / "matrix_state.yaml"


def tracking_dir(runtime_paths: RuntimePaths) -> Path:
    """Return the tracking directory for one runtime context."""
    return runtime_paths.storage_root / "tracking"


def encryption_keys_dir(runtime_paths: RuntimePaths) -> Path:
    """Return the encryption-keys directory for one runtime context."""
    return runtime_paths.storage_root / "encryption_keys"


def config_relative_path(
    raw_path: str | Path,
    runtime_paths: RuntimePaths,
) -> Path:
    """Return one configured path relative to the runtime config directory without resolving symlinks.

    Config-relative paths may use `${MINDROOM_STORAGE_PATH}` or
    `${MINDROOM_CONFIG_PATH}` placeholders only.
    """
    unresolved = Path(_expand_runtime_path_vars(os.fspath(raw_path), runtime_paths)).expanduser()
    if unresolved.is_absolute():
        return unresolved
    return runtime_paths.config_dir / unresolved


def resolve_config_relative_path(
    raw_path: str | Path,
    runtime_paths: RuntimePaths,
) -> Path:
    """Resolve a configured path, treating relative values as config-directory-relative."""
    return config_relative_path(raw_path, runtime_paths).resolve()


def resolve_config_relative_path_preserving_leaf(
    raw_path: str | Path,
    runtime_paths: RuntimePaths,
) -> Path:
    """Resolve a configured path lexically without following the final component."""
    unresolved = Path(_expand_runtime_path_vars(os.fspath(raw_path), runtime_paths)).expanduser()
    if unresolved.is_absolute():
        return unresolved
    return runtime_paths.config_dir / unresolved


def _docker_container_enabled(runtime_paths: RuntimePaths) -> bool:
    """Return whether MindRoom is running from the packaged container image."""
    return runtime_paths.env_flag("DOCKER_CONTAINER")


def _use_storage_path_for_workspace_assets(runtime_paths: RuntimePaths) -> bool:
    """Return whether writable workspace assets should live under persistent storage."""
    if not _docker_container_enabled(runtime_paths):
        return False
    configured_config_path = runtime_paths.process_env.get("MINDROOM_CONFIG_PATH")
    configured_storage_path = runtime_paths.process_env.get("MINDROOM_STORAGE_PATH")
    if configured_config_path is None or configured_storage_path is None:
        return False
    return (
        Path(configured_config_path).expanduser().resolve() == runtime_paths.config_path
        and Path(configured_storage_path).expanduser().resolve() == runtime_paths.storage_root
    )


def _avatars_dir(runtime_paths: RuntimePaths) -> Path:
    """Return the writable avatars directory for the active workspace.

    Source checkouts keep avatars next to the active config file so generated
    assets can be committed with the workspace.
    Containerized deployments usually mount `config.yaml` as a single file, so
    config-adjacent writes would be ephemeral; in that case, store writable
    overrides under the persistent MindRoom storage root instead.
    """
    if _use_storage_path_for_workspace_assets(runtime_paths):
        return runtime_paths.storage_root / "avatars"
    return runtime_paths.config_dir / "avatars"


def _bundled_avatars_dir() -> Path:
    """Return the bundled avatar directory shipped with a source checkout or runtime image."""
    return Path(__file__).resolve().parents[2] / "avatars"


def workspace_avatar_path(
    entity_type: str,
    entity_name: str,
    runtime_paths: RuntimePaths,
) -> Path:
    """Return the writable workspace avatar path for a managed entity."""
    return _avatars_dir(runtime_paths) / entity_type / f"{entity_name}.png"


def resolve_avatar_path(
    entity_type: str,
    entity_name: str,
    runtime_paths: RuntimePaths,
) -> Path:
    """Return the best available avatar path for a managed entity.

    Prefer a writable workspace override.
    Fall back to the bundled runtime assets when no workspace file exists yet.
    If neither exists, return the intended workspace path so callers that write
    new avatars know where to place them.
    """
    workspace_path = workspace_avatar_path(
        entity_type,
        entity_name,
        runtime_paths,
    )
    if workspace_path.exists():
        return workspace_path

    bundled_path = _bundled_avatars_dir() / entity_type / f"{entity_name}.png"
    if bundled_path.exists():
        return bundled_path

    return workspace_path


def _find_config(*, process_env: Mapping[str, str]) -> Path:
    """Find the first existing config file, or fall back to ~/.mindroom/config.yaml.

    Returns the original (possibly relative) path, not a resolved one,
    so CLI-facing defaults still display cleanly.
    """
    if configured_path := _configured_config_path(process_env):
        return configured_path
    for path in _CONFIG_SEARCH_PATHS:
        if path.exists():
            return path
    return _CONFIG_SEARCH_PATHS[-1]  # default to ~/.mindroom/config.yaml for creation


# Other constants
VOICE_PREFIX = "🎤 "
ORIGINAL_SENDER_KEY = "com.mindroom.original_sender"
SOURCE_KIND_KEY = "com.mindroom.source_kind"
PER_FIRE_THREAD_ROOT_KEY = "com.mindroom.per_fire_thread_root"
PER_FIRE_THREAD_ROOT_EVENT_ID_KEY = "com.mindroom.per_fire_thread_root_event_id"
SCHEDULED_HISTORY_LIMIT_KEY = "com.mindroom.history_limit"
HOOK_SOURCE_KEY = "com.mindroom.hook_source"
HOOK_MESSAGE_RECEIVED_DEPTH_KEY = "com.mindroom.message_received_depth"
SKIP_MENTIONS_KEY = "com.mindroom.skip_mentions"
VOICE_RAW_AUDIO_FALLBACK_KEY = "com.mindroom.voice_raw_audio_fallback"
VOICE_TRANSCRIPT_KEY = "com.mindroom.voice_transcript"
ATTACHMENT_IDS_KEY = "com.mindroom.attachment_ids"
AI_RUN_METADATA_KEY = "io.mindroom.ai_run"
MATRIX_TURN_SCHEMA_VERSION_METADATA_KEY = "matrix_turn_schema_version"
MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY = "matrix_turn_discovery_event_ids"
MATRIX_TURN_REDACTED_SOURCE_EVENT_IDS_METADATA_KEY = "matrix_turn_redacted_source_event_ids"
MATRIX_EVENT_ID_METADATA_KEY = "matrix_event_id"
MATRIX_MESSAGE_TARGET_ENRICHMENT_KEY = "matrix_message_target"
MATRIX_RESPONSE_EVENT_ID_METADATA_KEY = "matrix_response_event_id"
MATRIX_RESPONSE_OWNER_METADATA_KEY = "matrix_response_owner"
MATRIX_SEEN_EVENT_IDS_METADATA_KEY = "matrix_seen_event_ids"
MATRIX_HISTORY_SCOPE_METADATA_KEY = "matrix_history_scope"
MATRIX_CONVERSATION_TARGET_METADATA_KEY = "matrix_conversation_target"
MATRIX_SOURCE_EVENT_IDS_METADATA_KEY = "matrix_source_event_ids"
MATRIX_SOURCE_EVENT_PROMPTS_METADATA_KEY = "matrix_source_event_prompts"
MATRIX_SOURCE_EVENT_REVISIONS_METADATA_KEY = "matrix_source_event_revisions"
MATRIX_SOURCE_EVENT_METADATA_KEY = "matrix_source_event_metadata"
MINDROOM_COMPACTION_METADATA_KEY = "mindroom_compaction"
MINDROOM_MATRIX_HISTORY_METADATA_KEY = "mindroom_matrix_history"
COMPACTION_NOTICE_CONTENT_KEY = "io.mindroom.compaction"
STREAM_STATUS_KEY = "io.mindroom.stream_status"
STREAM_VISIBLE_BODY_KEY = "io.mindroom.visible_body"
STREAM_WARMUP_SUFFIX_KEY = "io.mindroom.warmup_suffix"
TOOL_TRACE_CONTENT_KEY = "io.mindroom.tool_trace"
STREAM_STATUS_PENDING = "pending"
STREAM_STATUS_STREAMING = "streaming"
STREAM_STATUS_COMPLETED = "completed"
STREAM_STATUS_CANCELLED = "cancelled"
STREAM_STATUS_INTERRUPTED = "interrupted"
STREAM_STATUS_ERROR = "error"

# Placeholder used in starter config templates. `mindroom connect` can
# automatically replace this token with the owner Matrix user ID returned
# by the provisioning service.
OWNER_MATRIX_USER_ID_PLACEHOLDER = "__MINDROOM_OWNER_USER_ID_FROM_PAIRING__"
OWNER_MATRIX_USER_ID_ENV = "MINDROOM_OWNER_USER_ID"


# Canonical mapping from provider name to the environment variable it requires.
# Other modules derive their own views from this single source of truth.
PROVIDER_ENV_KEYS: dict[str, str] = {
    "anthropic": "ANTHROPIC_API_KEY",
    "azure": runtime_env_policy.AZURE_OPENAI_ENV_BY_KEY["api_key"],
    "openai": "OPENAI_API_KEY",
    "google": "GOOGLE_API_KEY",
    "openrouter": "OPENROUTER_API_KEY",
    "deepseek": "DEEPSEEK_API_KEY",
    "cerebras": "CEREBRAS_API_KEY",
    "groq": "GROQ_API_KEY",
    "zai": "ZAI_API_KEY",
    "ollama": "OLLAMA_HOST",
}
# Dedicated workers start with no mirrored/shared credentials by default.
# Any service exposure into an isolated worker runtime must be explicitly authored.
DEFAULT_WORKER_GRANTABLE_CREDENTIALS = frozenset()

_CHROMADB_PY314_PATCHED = False


def env_key_for_provider(provider: str) -> str | None:
    """Get the environment variable name for a provider's API key.

    Handles the gemini→google alias so callers don't need to.
    """
    if provider == "gemini":
        return PROVIDER_ENV_KEYS.get("google")
    return PROVIDER_ENV_KEYS.get(provider)


def patch_chromadb_for_python314() -> None:
    """Patch pydantic internals so chromadb works on Python 3.14+.

    chromadb currently relies on pydantic v1 `BaseSettings` behavior and defines
    untyped fields in its settings model. This runtime shim can be removed once
    chromadb ships an upstream fix.
    """
    global _CHROMADB_PY314_PATCHED
    if _CHROMADB_PY314_PATCHED or sys.version_info < (3, 14):
        return

    import pydantic  # noqa: PLC0415
    from pydantic._internal import _model_construction  # noqa: PLC0415
    from pydantic_settings import BaseSettings  # noqa: PLC0415

    # pydantic-settings v2 defaults to extra="forbid", but pydantic v1's
    # BaseSettings silently ignored env vars / .env keys that didn't match
    # any field.  chromadb relies on that tolerance, so we must restore it.
    class _PermissiveBaseSettings(BaseSettings):
        model_config = BaseSettings.model_config.copy()
        model_config["extra"] = "ignore"

    pydantic.BaseSettings = _PermissiveBaseSettings

    original_inspect_namespace = _model_construction.inspect_namespace

    def _patched_inspect_namespace(*args: object, **kwargs: object) -> object:
        try:
            return original_inspect_namespace(*args, **kwargs)
        except pydantic.errors.PydanticUserError as exc:
            if "non-annotated attribute" not in str(exc):
                raise

            namespace = args[0] if args else kwargs.get("namespace")
            raw_annotations = args[1] if len(args) > 1 else kwargs.get("raw_annotations")
            if not isinstance(namespace, dict) or not isinstance(raw_annotations, dict):
                raise
            namespace_dict = cast("dict[str, object]", namespace)
            raw_annotations_dict = cast("dict[str, object]", raw_annotations)

            for field in (
                "chroma_coordinator_host",
                "chroma_logservice_host",
                "chroma_logservice_port",
            ):
                if field in namespace_dict and field not in raw_annotations_dict:
                    raw_annotations_dict[field] = type(namespace_dict[field])
            return original_inspect_namespace(*args, **kwargs)

    _model_construction.inspect_namespace = _patched_inspect_namespace
    _CHROMADB_PY314_PATCHED = True


def safe_replace(tmp_path: Path, target_path: Path) -> None:
    """Replace *target_path* with *tmp_path*, with a fallback for bind mounts.

    ``Path.replace`` performs an atomic rename which fails on some filesystems
    (e.g. Docker bind mounts) with ``OSError: [Errno 16] Device or resource
    busy``.  When that happens we fall back to a non-atomic copy.
    """
    try:
        tmp_path.replace(target_path)
    except OSError:
        shutil.copy2(tmp_path, target_path)
        target_fd = os.open(target_path, os.O_RDONLY)
        try:
            os.fsync(target_fd)
        finally:
            os.close(target_fd)
        tmp_path.unlink(missing_ok=True)


def ensure_writable_config_path(
    *,
    create_minimal: bool = False,
    runtime_paths: RuntimePaths,
) -> bool:
    """Ensure the writable config path exists when running from a managed template.

    Returns whether a config file exists after the call.
    """
    config_path = runtime_paths.config_path
    config_path.parent.mkdir(parents=True, exist_ok=True)

    if config_path.exists():
        return True

    template_path = runtime_env_path(runtime_paths, "MINDROOM_CONFIG_TEMPLATE") or config_path
    if template_path != config_path and template_path.exists():
        shutil.copyfile(template_path, config_path)
        config_path.chmod(0o600)
        print(f"Seeded config from template {template_path} -> {config_path}")
        return True

    if not create_minimal:
        return False

    config_path.write_text("agents: {}\nmodels: {}\n", encoding="utf-8")
    config_path.chmod(0o600)
    print(f"Created new config file at {config_path}")
    return True
