"""Generic worker-routing primitives for tool execution."""

from __future__ import annotations

import hashlib
import re
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal, cast

from mindroom.tool_system.context_bound_streams import context_bound_async_stream

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable, Iterator

    from mindroom.constants import RuntimePaths

WorkerScope = Literal["shared", "user", "user_agent"]
ResolvedWorkerKeyScope = Literal["shared", "user", "user_agent", "unscoped"]
_ExecutionChannel = Literal["matrix", "openai_compat"]

_WORKER_DIRNAME_MAX_PREFIX_LENGTH = 80
_DEFAULT_WORKER_NAME_PREFIX = "mindroom-worker"
_DNS_LABEL_MAX_LENGTH = 63
_WORKER_ID_DIGEST_LENGTH = 24
_DESCRIPTIVE_WORKER_ID_DIGEST_LENGTH = 10
_AGENT_WORKSPACE_DIRNAME = "workspace"
_PRIVATE_INSTANCE_ROOT_DIRNAME = "private_instances"
_SHARED_ONLY_INTEGRATION_NAMES = frozenset(
    {
        "spotify",
        "homeassistant",
    },
)
_LOCAL_ONLY_SHARED_INTEGRATION_TOOL_NAMES = frozenset(
    {
        "approved_egress",
        "attachments",
        "callback_manager",
        "desktop",
        "external_trigger_manager",
        "gmail",
        "google_calendar",
        "google_docs",
        "google_drive",
        "google_sheets",
        "homeassistant",
        "todo",
    },
)


@dataclass(frozen=True)
class ToolExecutionIdentity:
    """Serializable execution identity used for worker resolution."""

    channel: _ExecutionChannel
    agent_name: str
    requester_id: str | None
    room_id: str | None
    thread_id: str | None
    resolved_thread_id: str | None
    session_id: str | None
    tenant_id: str | None = None
    account_id: str | None = None
    transport_agent_name: str | None = None


type SerializedToolExecutionIdentity = dict[str, object]


_TOOL_EXECUTION_IDENTITY_OPTIONAL_PAYLOAD_FIELDS = (
    "requester_id",
    "room_id",
    "thread_id",
    "resolved_thread_id",
    "session_id",
    "tenant_id",
    "account_id",
    "transport_agent_name",
)


def serialize_tool_execution_identity(
    identity: ToolExecutionIdentity,
    *,
    include_transport_agent_name: bool = True,
) -> SerializedToolExecutionIdentity:
    """Return a JSON-safe payload for one execution identity."""
    payload = cast("SerializedToolExecutionIdentity", asdict(identity))
    if not include_transport_agent_name:
        payload.pop("transport_agent_name", None)
    return payload


def parse_tool_execution_identity_payload(
    payload: object,
    *,
    strict: bool = True,
    error_prefix: str = "Tool execution_identity",
) -> ToolExecutionIdentity | None:
    """Parse one JSON execution-identity payload with strict or lenient error handling."""
    if not isinstance(payload, dict):
        return _invalid_tool_execution_identity_payload(strict, f"{error_prefix} must be an object")

    raw_payload = cast("dict[str, object]", payload)
    channel = raw_payload.get("channel")
    if channel not in ("matrix", "openai_compat"):
        return _invalid_tool_execution_identity_payload(
            strict,
            f"{error_prefix}.channel must be matrix or openai_compat",
        )

    agent_name = raw_payload.get("agent_name")
    if not isinstance(agent_name, str) or not agent_name.strip():
        return _invalid_tool_execution_identity_payload(
            strict,
            f"{error_prefix}.agent_name must be a non-empty string",
        )

    optional_values: dict[str, str | None] = {}
    for field_name in _TOOL_EXECUTION_IDENTITY_OPTIONAL_PAYLOAD_FIELDS:
        value = raw_payload.get(field_name)
        if value is not None and not isinstance(value, str):
            return _invalid_tool_execution_identity_payload(
                strict,
                f"{error_prefix}.{field_name} must be a string when present",
            )
        optional_values[field_name] = value

    return ToolExecutionIdentity(
        channel=cast("_ExecutionChannel", channel),
        agent_name=agent_name,
        requester_id=optional_values["requester_id"],
        room_id=optional_values["room_id"],
        thread_id=optional_values["thread_id"],
        resolved_thread_id=optional_values["resolved_thread_id"],
        session_id=optional_values["session_id"],
        tenant_id=optional_values["tenant_id"],
        account_id=optional_values["account_id"],
        transport_agent_name=optional_values["transport_agent_name"],
    )


def _invalid_tool_execution_identity_payload(strict: bool, message: str) -> None:
    if strict:
        raise TypeError(message)


@dataclass(frozen=True)
class _ResolvedWorkerExecution:
    """Resolved worker execution scope from explicit worker-scope policy."""

    worker_scope: WorkerScope | None
    execution_identity: ToolExecutionIdentity | None
    worker_key: str | None


@dataclass(frozen=True)
class ResolvedWorkerTarget:
    """Resolved worker target carried through tool construction and sandbox routing.

    This layer still uses `worker_scope` because it is about worker reuse/routing.
    Private agents reach this seam through the already-derived execution scope from
    `private.per`; do not read this field as the raw authored agent config.
    """

    worker_scope: WorkerScope | None
    routing_agent_name: str | None
    execution_identity: ToolExecutionIdentity | None
    tenant_id: str | None
    account_id: str | None
    worker_key: str | None
    private_agent_names: frozenset[str] | None = None


_TOOL_EXECUTION_IDENTITY: ContextVar[ToolExecutionIdentity | None] = ContextVar(
    "tool_execution_identity",
    default=None,
)


def get_tool_execution_identity() -> ToolExecutionIdentity | None:
    """Return the current tool execution identity."""
    return _TOOL_EXECUTION_IDENTITY.get()


def active_tool_execution_identity(
    execution_identity: ToolExecutionIdentity | None,
) -> ToolExecutionIdentity | None:
    """Return the explicit execution identity or the active boundary context."""
    if execution_identity is not None:
        return execution_identity
    return get_tool_execution_identity()


@contextmanager
def tool_execution_identity(identity: ToolExecutionIdentity | None) -> Iterator[None]:
    """Set the current tool execution identity for the active execution scope."""
    token = _TOOL_EXECUTION_IDENTITY.set(identity)
    try:
        yield
    finally:
        _TOOL_EXECUTION_IDENTITY.reset(token)


async def run_with_tool_execution_identity[ReturnT](
    identity: ToolExecutionIdentity | None,
    *,
    operation: Callable[[], Awaitable[ReturnT]],
) -> ReturnT:
    """Execute one async operation inside one execution-identity boundary."""
    with tool_execution_identity(identity):
        return await operation()


def stream_with_tool_execution_identity[ChunkT](
    identity: ToolExecutionIdentity | None,
    *,
    stream_factory: Callable[[], AsyncIterator[ChunkT]],
) -> AsyncIterator[ChunkT]:
    """Wrap one async iterator without spanning execution-identity tokens across yields."""
    return context_bound_async_stream(
        context_factory=lambda: tool_execution_identity(identity),
        stream_factory=stream_factory,
    )


def normalize_worker_key_part(value: str) -> str:
    """Return one normalized worker-key component."""
    normalized = re.sub(r"[^a-zA-Z0-9._@+-]+", "_", value.strip()).strip("_")
    return normalized or "default"


def _digest_and_safe_prefix(worker_key: str, prefix: str, *, digest_length: int) -> tuple[str, str]:
    """Return the worker-key digest and a prefix normalized to a DNS-label-safe slug beside it."""
    digest = hashlib.sha256(worker_key.encode("utf-8")).hexdigest()[:digest_length]
    normalized_prefix = re.sub(r"[^a-z0-9-]+", "-", prefix.lower()).strip("-") or _DEFAULT_WORKER_NAME_PREFIX
    max_prefix_length = _DNS_LABEL_MAX_LENGTH - len(digest) - 1
    safe_prefix = normalized_prefix[:max_prefix_length].rstrip("-")
    if not safe_prefix:
        safe_prefix = _DEFAULT_WORKER_NAME_PREFIX[:max_prefix_length].rstrip("-") or "worker"
    return digest, safe_prefix


def worker_id_for_key(worker_key: str, *, prefix: str) -> str:
    """Return a DNS-safe resource name for one worker key (63-char label limit)."""
    digest, safe_prefix = _digest_and_safe_prefix(worker_key, prefix, digest_length=_WORKER_ID_DIGEST_LENGTH)
    return f"{safe_prefix}-{digest}"


def descriptive_worker_id_for_key(worker_key: str, *, prefix: str) -> str:
    """Return a DNS-safe resource name that keeps the worker key's scope readable.

    Embeds the key's tenant/scope/requester/agent parts as a lowercase slug
    (dropping the `v1` version tag and the `default` tenant, and reducing
    Matrix requesters to their localpart) so listings stay human-readable,
    with a short digest for uniqueness after truncation.
    """
    digest, safe_prefix = _digest_and_safe_prefix(
        worker_key,
        prefix,
        digest_length=_DESCRIPTIVE_WORKER_ID_DIGEST_LENGTH,
    )
    max_slug_length = max(0, _DNS_LABEL_MAX_LENGTH - len(safe_prefix) - len(digest) - 2)
    slug = _worker_key_slug(worker_key, max_length=max_slug_length)
    if not slug:
        return f"{safe_prefix}-{digest}"
    return f"{safe_prefix}-{slug}-{digest}"


def _worker_key_slug(worker_key: str, *, max_length: int) -> str:
    parts = worker_key.split(":")
    if len(parts) >= 4 and parts[0] == "v1":
        tenant, scope, rest = parts[1], parts[2], parts[3:]
        if scope == "user":
            rest = [_requester_localpart(":".join(rest))]
        elif scope == "user_agent" and len(rest) >= 2:
            rest = [_requester_localpart(":".join(rest[:-1])), rest[-1]]
        parts = ([] if tenant == "default" else [tenant]) + [scope, *rest]
    slug = re.sub(r"[^a-z0-9]+", "-", ":".join(parts).lower()).strip("-")
    return slug[:max_length].rstrip("-")


def _requester_localpart(requester: str) -> str:
    """Return the localpart of a Matrix-style requester; other requesters pass through."""
    if requester.startswith("@"):
        return requester[1:].split(":", 1)[0]
    return requester


def _normalize_worker_requester_part(value: str) -> str:
    normalized = re.sub(r"[^a-zA-Z0-9._:@+-]+", "_", value.strip()).strip("_")
    return normalized or "default"


def _normalize_worker_dir_part(value: str) -> str:
    normalized = re.sub(r"[^a-zA-Z0-9._@+-]+", "_", value.strip()).strip("_")
    return normalized or "worker"


def _identity_requester_key(identity: ToolExecutionIdentity) -> str | None:
    if identity.requester_id:
        return _normalize_worker_requester_part(identity.requester_id)
    return None


def build_tool_execution_identity(
    *,
    channel: _ExecutionChannel,
    agent_name: str,
    transport_agent_name: str | None = None,
    runtime_paths: RuntimePaths,
    requester_id: str | None,
    room_id: str | None,
    thread_id: str | None,
    resolved_thread_id: str | None,
    session_id: str | None,
) -> ToolExecutionIdentity:
    """Build the ingress execution identity for one request or shared materialization."""
    return ToolExecutionIdentity(
        channel=channel,
        agent_name=agent_name,
        transport_agent_name=transport_agent_name or agent_name,
        requester_id=requester_id,
        room_id=room_id,
        thread_id=thread_id,
        resolved_thread_id=resolved_thread_id,
        session_id=session_id,
        tenant_id=runtime_paths.env_value("CUSTOMER_ID"),
        account_id=runtime_paths.env_value("ACCOUNT_ID"),
    )


def resolve_worker_execution_scope(
    worker_scope: WorkerScope | None,
    execution_identity: ToolExecutionIdentity | None,
    *,
    agent_name: str | None = None,
    tenant_id: str | None = None,
    account_id: str | None = None,
) -> _ResolvedWorkerExecution:
    """Resolve worker execution identity and key from explicit scope inputs."""
    resolved_execution_identity = _resolve_execution_identity_for_worker_scope(
        worker_scope,
        agent_name=agent_name,
        execution_identity=execution_identity,
        tenant_id=tenant_id,
        account_id=account_id,
    )
    worker_key: str | None = None
    if worker_scope is not None and resolved_execution_identity is not None:
        worker_key = resolve_worker_key(
            worker_scope,
            resolved_execution_identity,
            agent_name=agent_name,
        )
    return _ResolvedWorkerExecution(
        worker_scope=worker_scope,
        execution_identity=resolved_execution_identity,
        worker_key=worker_key,
    )


def resolve_worker_target(
    worker_scope: WorkerScope | None,
    routing_agent_name: str | None,
    execution_identity: ToolExecutionIdentity | None,
    *,
    tenant_id: str | None = None,
    account_id: str | None = None,
    private_agent_names: frozenset[str] | None = None,
) -> ResolvedWorkerTarget:
    """Resolve one explicit worker target for tool construction and sandbox routing."""
    effective_agent_name = routing_agent_name
    if effective_agent_name is None and execution_identity is not None:
        effective_agent_name = execution_identity.agent_name

    resolved_worker_execution = resolve_worker_execution_scope(
        worker_scope,
        execution_identity=execution_identity,
        agent_name=effective_agent_name,
        tenant_id=tenant_id,
        account_id=account_id,
    )
    resolved_execution_identity = resolved_worker_execution.execution_identity
    return ResolvedWorkerTarget(
        worker_scope=worker_scope,
        routing_agent_name=effective_agent_name,
        execution_identity=resolved_execution_identity,
        tenant_id=(
            tenant_id
            or (
                resolved_execution_identity.tenant_id
                if resolved_execution_identity is not None and resolved_execution_identity.tenant_id is not None
                else None
            )
        ),
        account_id=(
            account_id
            or (
                resolved_execution_identity.account_id
                if resolved_execution_identity is not None and resolved_execution_identity.account_id is not None
                else None
            )
        ),
        worker_key=resolved_worker_execution.worker_key,
        private_agent_names=private_agent_names,
    )


def build_worker_target_from_runtime_env(
    worker_scope: WorkerScope | None,
    routing_agent_name: str | None,
    execution_identity: ToolExecutionIdentity | None,
    runtime_paths: RuntimePaths,
    *,
    private_agent_names: frozenset[str] | None = None,
) -> ResolvedWorkerTarget:
    """Build one worker target at the ingress boundary from runtime tenant/account env."""
    return resolve_worker_target(
        worker_scope,
        routing_agent_name,
        execution_identity=execution_identity,
        tenant_id=runtime_paths.env_value("CUSTOMER_ID"),
        account_id=runtime_paths.env_value("ACCOUNT_ID"),
        private_agent_names=private_agent_names,
    )


def build_agent_toolkit_worker_target(
    worker_scope: WorkerScope | None,
    agent_name: str,
    *,
    is_private: bool,
    execution_identity: ToolExecutionIdentity | None,
    runtime_paths: RuntimePaths,
) -> ResolvedWorkerTarget:
    """Build the worker target used when constructing one agent's registered toolkits.

    Single source of truth for the `private_agent_names` derivation shared by
    agent toolkit construction and dispatch-time tool composition.
    """
    if worker_scope == "user_agent":
        private_agent_names = frozenset({agent_name}) if is_private else frozenset()
    else:
        private_agent_names = None
    return build_worker_target_from_runtime_env(
        worker_scope,
        agent_name,
        execution_identity=execution_identity,
        runtime_paths=runtime_paths,
        private_agent_names=private_agent_names,
    )


def worker_scope_allows_shared_only_integrations(worker_scope: WorkerScope | None) -> bool:
    """Return whether a worker scope can use shared-only dashboard integrations."""
    return worker_scope in (None, "shared")


def _requires_shared_only_integration_scope(name: str) -> bool:
    """Return whether a tool or dashboard integration is restricted to shared scope.

    MCP registry tools are supported on every scope: OAuth-backed servers use
    requester-scoped sessions, and non-OAuth servers always execute through the
    shared server session without requester credentials.
    """
    return name in _SHARED_ONLY_INTEGRATION_NAMES


def supports_tool_name_for_worker_scope(name: str, worker_scope: WorkerScope | None) -> bool:
    """Return whether one tool/integration name is supported for the effective execution scope."""
    return not _requires_shared_only_integration_scope(name) or worker_scope_allows_shared_only_integrations(
        worker_scope,
    )


def unsupported_shared_only_integration_names(
    names: list[str],
    worker_scope: WorkerScope | None,
) -> list[str]:
    """Return shared-only integration names that are invalid for the effective execution scope."""
    if worker_scope_allows_shared_only_integrations(worker_scope):
        return []
    return [name for name in names if _requires_shared_only_integration_scope(name)]


def tool_stays_local(name: str) -> bool:
    """Return whether one integration tool always stays in the primary runtime."""
    return name in _LOCAL_ONLY_SHARED_INTEGRATION_TOOL_NAMES


def unsupported_shared_only_integration_message(
    name: str,
    worker_scope: WorkerScope | None,
    *,
    agent_name: str | None = None,
    subject: str = "Integration",
    scope_label: str | None = None,
) -> str:
    """Return the user-facing error for shared-only integrations on isolating scopes."""
    resolved_scope_label = scope_label or (
        f"execution_scope={worker_scope}" if worker_scope is not None else "unscoped"
    )
    agent_detail = f"Agent '{agent_name}' uses " if agent_name else "This request uses "
    return (
        f"{subject} '{name}' is only supported for shared deployment credentials or agents with "
        f"worker_scope=shared. {agent_detail}{resolved_scope_label}."
    )


def resolve_worker_key(
    worker_scope: WorkerScope,
    identity: ToolExecutionIdentity,
    *,
    agent_name: str | None = None,
) -> str | None:
    """Derive a stable worker key from scope and execution identity."""
    tenant_key = normalize_worker_key_part(identity.tenant_id or identity.account_id or "default")
    effective_agent_name = normalize_worker_key_part(agent_name or identity.agent_name)
    worker_key: str | None

    if worker_scope == "shared":
        worker_key = f"v1:{tenant_key}:shared:{effective_agent_name}"
    elif worker_scope == "user":
        requester_key = _identity_requester_key(identity)
        if requester_key is None:
            return None
        worker_key = f"v1:{tenant_key}:user:{requester_key}"
    elif worker_scope == "user_agent":
        requester_key = _identity_requester_key(identity)
        if requester_key is None:
            return None
        worker_key = f"v1:{tenant_key}:user_agent:{requester_key}:{effective_agent_name}"
    else:
        msg = f"Unknown worker scope: {worker_scope}"
        raise ValueError(msg)

    return worker_key


def resolve_unscoped_worker_key(
    agent_name: str,
    execution_identity: ToolExecutionIdentity | None = None,
    *,
    tenant_id: str | None = None,
    account_id: str | None = None,
) -> str:
    """Derive a stable backend worker key for unscoped sandbox execution."""
    identity = execution_identity
    tenant_key = normalize_worker_key_part(
        tenant_id
        or (identity.tenant_id if identity is not None and identity.tenant_id is not None else None)
        or account_id
        or (identity.account_id if identity is not None and identity.account_id is not None else None)
        or "default",
    )
    effective_agent_name = normalize_worker_key_part(agent_name)
    return f"v1:{tenant_key}:unscoped:{effective_agent_name}"


def require_worker_key_for_scope(
    worker_scope: WorkerScope,
    execution_identity: ToolExecutionIdentity | None,
    *,
    agent_name: str | None = None,
    tenant_id: str | None = None,
    account_id: str | None = None,
    failure_message: str,
) -> str:
    """Resolve one worker key from explicit inputs or raise with a caller-owned message."""
    worker_key = resolve_worker_execution_scope(
        worker_scope,
        execution_identity=execution_identity,
        agent_name=agent_name,
        tenant_id=tenant_id,
        account_id=account_id,
    ).worker_key
    if worker_key is None:
        raise ValueError(failure_message)
    return worker_key


def resolved_worker_key_scope(worker_key: str) -> ResolvedWorkerKeyScope | None:
    """Return the parsed scope discriminator for one resolved worker key."""
    parts = worker_key.split(":")
    if len(parts) < 4 or parts[0] != "v1":
        return None
    scope = parts[2]
    if scope not in {"shared", "user", "user_agent", "unscoped"}:
        return None
    return cast("ResolvedWorkerKeyScope", scope)


def requires_explicit_private_agent_visibility(worker_key: str) -> bool:
    """Return whether visible private-agent names must be supplied for this worker."""
    return resolved_worker_key_scope(worker_key) == "user_agent"


def worker_key_agent_name(worker_key: str) -> str | None:
    """Return the encoded agent name for one resolved worker key, when present."""
    scope = resolved_worker_key_scope(worker_key)
    if scope is None or scope == "user":
        return None

    parts = worker_key.split(":")
    min_parts_by_scope = {
        "shared": 4,
        "unscoped": 4,
        "user_agent": 5,
    }
    min_parts = min_parts_by_scope.get(scope)
    if min_parts is None or len(parts) < min_parts:
        return None
    return parts[3] if scope in {"shared", "unscoped"} else parts[-1]


def _resolve_execution_identity_for_worker_scope(
    worker_scope: WorkerScope | None,
    execution_identity: ToolExecutionIdentity | None = None,
    *,
    agent_name: str | None = None,
    tenant_id: str | None = None,
    account_id: str | None = None,
) -> ToolExecutionIdentity | None:
    """Resolve the execution identity used for worker scope decisions.

    Shared-scope state can be resolved from agent identity plus tenant/account
    even when no live request context exists yet. Isolating scopes still
    require an explicit execution identity.
    """
    if execution_identity is not None:
        return execution_identity

    if worker_scope != "shared" or agent_name is None:
        return None

    if tenant_id is None and account_id is None:
        return None

    return ToolExecutionIdentity(
        channel="matrix",
        agent_name=agent_name,
        requester_id=None,
        room_id=None,
        thread_id=None,
        resolved_thread_id=None,
        session_id=None,
        tenant_id=tenant_id,
        account_id=account_id,
    )


def worker_dir_name(worker_key: str) -> str:
    """Return a stable filesystem-safe dirname for a worker key."""
    prefix = _normalize_worker_dir_part(worker_key)
    prefix = prefix[:_WORKER_DIRNAME_MAX_PREFIX_LENGTH].rstrip("._-")
    if not prefix:
        prefix = "worker"
    digest = hashlib.sha256(worker_key.encode("utf-8")).hexdigest()[:16]
    return f"{prefix}-{digest}"


def worker_root_path(base_storage_path: Path, worker_key: str) -> Path:
    """Return the persistent runtime root path for a worker key."""
    resolved_base_path = shared_storage_root(base_storage_path)
    if _is_resolved_worker_root(resolved_base_path, worker_key):
        return resolved_base_path
    return resolved_base_path / "workers" / worker_dir_name(worker_key)


def shared_storage_root(base_storage_path: Path) -> Path:
    """Return the canonical shared storage root.

    Callers must pass the actual shared storage root, not an `agents/<name>` or
    `workers/<name>` child path. Security-sensitive path checks should fail closed
    rather than guess by peeling path segments based only on directory names.
    """
    return base_storage_path.expanduser().resolve()


def agent_state_root_path(base_storage_path: Path, agent_name: str) -> Path:
    """Return the canonical shared state root for one agent.

    Agent-state resolution accepts the shared storage root or a pre-resolved
    canonical agent root.
    """
    resolved_base_path = shared_storage_root(base_storage_path)
    if _is_resolved_agent_state_root(resolved_base_path, agent_name):
        return resolved_base_path
    return resolved_base_path / "agents" / _normalize_worker_dir_part(agent_name)


def private_instance_scope_root_path(base_storage_path: Path, worker_key: str) -> Path:
    """Return the canonical shared root for one worker-scoped private-instance namespace."""
    resolved_base_path = shared_storage_root(base_storage_path)
    if _is_resolved_private_instance_scope_root(resolved_base_path, worker_key):
        return resolved_base_path
    return resolved_base_path / _PRIVATE_INSTANCE_ROOT_DIRNAME / worker_dir_name(worker_key)


def _private_instance_state_root_path(
    base_storage_path: Path,
    *,
    worker_key: str,
    agent_name: str,
) -> Path:
    """Return the canonical durable state root for one private agent instance."""
    return private_instance_scope_root_path(base_storage_path, worker_key) / _normalize_worker_dir_part(agent_name)


def private_instance_state_root_for_requester(
    base_storage_path: Path,
    *,
    requester_id: str,
    agent_name: str,
    worker_scope: WorkerScope,
    runtime_paths: RuntimePaths,
) -> Path | None:
    """Return the private-instance state root one requester would own for an agent.

    Instance directory names embed a one-way digest of the worker key, so ownership can only be
    established by forward-computing the path a known requester would get and comparing it against
    the directories that exist on disk.
    """
    identity = build_tool_execution_identity(
        channel="matrix",
        agent_name=agent_name,
        runtime_paths=runtime_paths,
        requester_id=requester_id,
        room_id=None,
        thread_id=None,
        resolved_thread_id=None,
        session_id=None,
    )
    worker_key = resolve_worker_key(worker_scope, identity, agent_name=agent_name)
    if worker_key is None:
        return None
    return _private_instance_state_root_path(base_storage_path, worker_key=worker_key, agent_name=agent_name)


def _is_resolved_agent_state_root(path: Path, agent_name: str) -> bool:
    resolved_path = path.expanduser().resolve()
    return resolved_path.parent.name == "agents" and resolved_path.name == _normalize_worker_dir_part(agent_name)


def _is_resolved_private_instance_scope_root(path: Path, worker_key: str) -> bool:
    resolved_path = path.expanduser().resolve()
    return resolved_path.parent.name == _PRIVATE_INSTANCE_ROOT_DIRNAME and resolved_path.name == worker_dir_name(
        worker_key,
    )


def _is_resolved_worker_root(path: Path, worker_key: str) -> bool:
    resolved_path = path.expanduser().resolve()
    return resolved_path.parent.name == "workers" and resolved_path.name == worker_dir_name(worker_key)


def visible_state_roots_for_worker_key(
    base_storage_path: Path,
    worker_key: str,
    *,
    private_agent_names: frozenset[str] = frozenset(),
) -> tuple[Path, ...]:
    """Return the canonical durable state roots a worker key is allowed to see by default.

    Shared agent roots remain canonical for normal agents.
    Private-instance roots live under a separate shared-storage namespace keyed by
    worker scope so they are durable without becoming worker-owned state.
    `user` intentionally sees the shared `agents/` tree plus its own
    private-instance namespace because it acts as a per-requester multi-agent
    workstation.
    """
    scope = resolved_worker_key_scope(worker_key)
    if scope is None:
        return ()
    if scope == "user":
        return (
            shared_storage_root(base_storage_path) / "agents",
            private_instance_scope_root_path(base_storage_path, worker_key),
        )

    agent_name = worker_key_agent_name(worker_key)
    if agent_name is None:
        return ()
    if scope == "user_agent" and agent_name in private_agent_names:
        return (
            _private_instance_state_root_path(
                base_storage_path,
                worker_key=worker_key,
                agent_name=agent_name,
            ),
        )
    return (agent_state_root_path(base_storage_path, agent_name),)


def agent_workspace_root_path(base_storage_path: Path, agent_name: str) -> Path:
    """Return the canonical workspace root for one agent."""
    return agent_state_root_path(base_storage_path, agent_name) / _AGENT_WORKSPACE_DIRNAME


def agent_workspace_relative_path(path_text: str) -> Path:
    """Validate and normalize a path that must live inside an agent workspace."""
    normalized_text = path_text.strip()
    if not normalized_text:
        msg = "Agent-owned paths must not be empty."
        raise ValueError(msg)
    if "$" in normalized_text:
        msg = f"Agent-owned paths must be workspace-relative literals, not env-variable references: {path_text}"
        raise ValueError(msg)

    candidate = Path(normalized_text).expanduser()
    if candidate.is_absolute():
        msg = f"Agent-owned paths must be workspace-relative, not absolute: {path_text}"
        raise ValueError(msg)

    if ".." in candidate.parts:
        msg = f"Agent-owned paths must stay within the agent workspace: {path_text}"
        raise ValueError(msg)

    return candidate


def _resolve_agent_workspace_target(relative_path: Path, *, agent_root: Path) -> Path:
    candidate = (agent_root / relative_path).resolve()
    resolved_root = agent_root.resolve()
    try:
        candidate.relative_to(resolved_root)
    except ValueError as exc:
        msg = f"Agent-owned paths must stay within {resolved_root}: {relative_path}"
        raise ValueError(msg) from exc
    return candidate


def resolve_agent_owned_path(
    path_text: str,
    *,
    agent_name: str,
    base_storage_path: Path,
) -> Path:
    """Resolve one agent-owned path into the canonical shared agent workspace.

    Durable agent files are shared per agent across all requesters and worker scopes.
    ``worker_scope`` only changes which runtime executes the tool call, not which
    files are authoritative.
    """
    relative_target = agent_workspace_relative_path(path_text)
    agent_workspace_root = agent_workspace_root_path(base_storage_path, agent_name).resolve()
    return _resolve_agent_workspace_target(relative_target, agent_root=agent_workspace_root)


def resolve_agent_state_storage_path(
    *,
    agent_name: str,
    base_storage_path: Path,
) -> Path:
    """Return the canonical durable state root for one agent.

    Requester-scoped worker runtimes do not partition file-backed memory, mem0 state,
    sessions, or learning. All durable agent state lives under one root per agent.
    """
    return agent_state_root_path(base_storage_path, agent_name)
